Redis

Redis sink connector

Description

The Redis sink connector writes upstream rows to Redis in batch or streaming jobs. It supports single-node Redis and Redis Cluster, and can write to key/string, hash, list, set, and zset data types.

The configured key can be either a literal Redis key or an upstream field name. When support_custom_key = true, the connector can build the Redis key from one or more upstream fields, for example user:${id}.

Support Those Engines

Spark
Flink
SeaTunnel Zeta

Key Features

• exactly-once
• cdc
• support multiple table write
• timer flush

Supported DataSource Info

To use the Redis connector, the following dependency is required. It can be installed by install-plugin.sh or downloaded from Maven Central.

Datasource	Dependency
Redis	Download

Options

Name	Type	Required	Default	Description
host	string	Yes when mode = SINGLE	-	Redis host for single-node mode.
port	int	No	6379	Redis port for single-node mode.
nodes	list	Yes when mode = CLUSTER	-	Redis Cluster nodes, for example ["redis-0:6379", "redis-1:6379"].
mode	string	No	SINGLE	Redis deployment mode. Supported values are SINGLE and CLUSTER.
user	string	No	-	Redis ACL username.
auth	string	No	-	Redis authentication password.
db_num	int	No	0	Redis database index. This option is used in single-node mode.
key	string	Yes	-	Redis key, upstream field name, or key template when support_custom_key = true.
data_type	string	Yes	-	Redis data type. Supported values are KEY, STRING, HASH, LIST, SET, and ZSET.
format	string	No	JSON	Serialization format used when no value field is configured. Supported values are JSON and TEXT.
field_delimiter	string	No	,	Field delimiter used when format = TEXT.
batch_size	int	No	10	Maximum number of rows buffered before one batch write.
expire	long	No	-1	Key expiration time in seconds. Values less than or equal to 0 mean no expiration is set.
support_custom_key	boolean	No	false	Whether to replace placeholders in key with upstream field values.
value_field	string	No	-	Upstream field used as the Redis value for KEY/STRING, LIST, SET, and ZSET.
hash_key_field	string	No	-	Upstream field used as the Redis hash field when data_type = HASH.
hash_value_field	string	No	-	Upstream field used as the Redis hash value when data_type = HASH.
multi_table_sink_replica	int	No	1	Writer replica count for multi-table writes.
common-options	config	No	-	Sink plugin common parameters. See Sink Common Options.

Write Rules

key

When support_custom_key = false, the connector first checks whether key matches an upstream field name:

• If the field exists, the field value is used as the Redis key.
• If the field does not exist, key itself is used as a fixed Redis key.

When support_custom_key = true, placeholders in key are replaced by upstream field values. Both ${field} and the legacy {field} style are supported.

data_type

• KEY and STRING: write one Redis string value for each row. Later rows overwrite earlier rows with the same key.
• HASH: write one or more fields into a Redis hash. Configure hash_key_field to choose the hash field name.
• LIST: append each row value to a Redis list.
• SET: add each row value to a Redis set.
• ZSET: add each row value to a Redis sorted set with score 1.

value_field

For KEY/STRING, LIST, SET, and ZSET, configure value_field when only one upstream field should be written as the Redis value. If value_field is not configured, the connector serializes the whole upstream row using format.

hash_key_field and hash_value_field

For HASH, hash_key_field chooses the Redis hash field. If hash_value_field is configured, that field value is written as the Redis hash value. If hash_value_field is not configured, the connector serializes the whole upstream row as the hash value.

multi_table_sink_replica

Replica count for multi-table sink writers. It applies when upstream rows carry table identifiers and the job writes multiple Redis tables in one pipeline.

For multi-table jobs, key may include ${table_name} so rows from different upstream tables are written to separate Redis keys, for example key = "redis-result-${table_name}".

Examples

Write Rows To A Redis List

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    schema = {
      fields {
        id = int
        name = string
      }
    }
    rows = [
      { kind = INSERT, fields = [1, "Alice"] },
      { kind = INSERT, fields = [2, "Bob"] }
    ]
  }
}

sink {
  Redis {
    host = "localhost"
    port = 6379
    key = "person_list"
    data_type = LIST
    value_field = "name"
  }
}

Use A Custom Key Template

sink {
  Redis {
    host = "localhost"
    port = 6379
    key = "person:${id}"
    support_custom_key = true
    data_type = KEY
    format = JSON
  }
}

Write Hash Fields

sink {
  Redis {
    host = "localhost"
    port = 6379
    key = "person_hash"
    data_type = HASH
    hash_key_field = "id"
    hash_value_field = "name"
  }
}

Write To Redis Cluster With Expiration

sink {
  Redis {
    mode = CLUSTER
    nodes = ["redis-cluster-0:6379", "redis-cluster-1:6379", "redis-cluster-2:6379"]
    key = "event:${id}"
    support_custom_key = true
    data_type = KEY
    value_field = "name"
    batch_size = 20
    expire = 30
  }
}

Changelog

Change Log

Change	Commit	Version
[Improve][Connector-V2] Use key_field_name option when reading Redis hash data (#9642)	https://github.com/apache/seatunnel/commit/5d214a7305	2.3.12
[Feature][Redis] Add redis key into the result record (#9574)	https://github.com/apache/seatunnel/commit/6e8b7c5da5	2.3.12
[Fix][Connector-Redis] Redis did not write successfully, but the task did not fail (#9055)	https://github.com/apache/seatunnel/commit/07510ed937	2.3.11
[hotfix][redis] fix npe cause by null host parameter (#8881)	https://github.com/apache/seatunnel/commit/7bd5865165	2.3.10
[Improve][Redis] Optimized Redis connection params (#8841)	https://github.com/apache/seatunnel/commit/e56f06cdf0	2.3.10
[Improve] restruct connector common options (#8634)	https://github.com/apache/seatunnel/commit/f3499a6eeb	2.3.10
[improve] update Redis connector config option (#8631)	https://github.com/apache/seatunnel/commit/f1c313eea6	2.3.10
[Feature][Redis] Flush data when the time reaches checkpoint.interval and update test case (#8308)	https://github.com/apache/seatunnel/commit/e15757bcd7	2.3.9
Revert "[Feature][Redis] Flush data when the time reaches checkpoint interval" and "[Feature][CDC] Add 'schema-changes.enabled' options" (#8278)	https://github.com/apache/seatunnel/commit/fcb2938286	2.3.9
[Feature][Redis] Flush data when the time reaches checkpoint.interval (#8198)	https://github.com/apache/seatunnel/commit/2e24941e6a	2.3.9
[Hotfix] Fix redis sink NPE (#8171)	https://github.com/apache/seatunnel/commit/6b9074e769	2.3.9
[Improve][dist]add shade check rule (#8136)	https://github.com/apache/seatunnel/commit/51ef800016	2.3.9
[Feature][Connector-Redis] Redis connector support delete data (#7994)	https://github.com/apache/seatunnel/commit/02a35c3979	2.3.9
[Improve][Connector-V2] Redis support custom key and value (#7888)	https://github.com/apache/seatunnel/commit/ef2c3c7283	2.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
[improve][Redis]Redis scan command supports versions 5, 6, 7 (#7666)	https://github.com/apache/seatunnel/commit/6e70cbe334	2.3.8
[Improve][Connector] Add multi-table sink option check (#7360)	https://github.com/apache/seatunnel/commit/2489f6446b	2.3.7
[Feature][Core] Support using upstream table placeholders in sink options and auto replacement (#7131)	https://github.com/apache/seatunnel/commit/c4ca74122c	2.3.6
[Improve][Redis] Redis reader use scan cammnd instead of keys, single mode reader/writer support batch (#7087)	https://github.com/apache/seatunnel/commit/be37f05c07	2.3.6
[Feature][Kafka] Support multi-table source read (#5992)	https://github.com/apache/seatunnel/commit/60104602d1	2.3.6
[Improve][Connector-V2]Support multi-table sink feature for redis (#6314)	https://github.com/apache/seatunnel/commit/fed89ae3fc	2.3.5
[Feature][Core] Upgrade flink source translation (#5100)	https://github.com/apache/seatunnel/commit/5aabb14a94	2.3.4
[Feature][Connector-V2] Support TableSourceFactory/TableSinkFactory on redis (#5901)	https://github.com/apache/seatunnel/commit/e84dcb8c10	2.3.4
[Improve][Common] Introduce new error define rule (#5793)	https://github.com/apache/seatunnel/commit/9d1b2582b2	2.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)	https://github.com/apache/seatunnel/commit/8de7408100	2.3.4
[Improve][Connector-v2][Redis] Redis support select db (#5570)	https://github.com/apache/seatunnel/commit/77fbbbd0ee	2.3.4
Support config column/primaryKey/constraintKey in schema (#5564)	https://github.com/apache/seatunnel/commit/eac76b4e50	2.3.4
[Feature][Connector-v2][RedisSink]Support redis to set expiration time. (#4975)	https://github.com/apache/seatunnel/commit/b5321ff1d2	2.3.3
Merge branch 'dev' into merge/cdc	https://github.com/apache/seatunnel/commit/4324ee1912	2.3.1
[Improve][Project] Code format with spotless plugin.	https://github.com/apache/seatunnel/commit/423b583038	2.3.1
[improve][api] Refactoring schema parse (#4157)	https://github.com/apache/seatunnel/commit/b2f573a13e	2.3.1
[Improve][build] Give the maven module a human readable name (#4114)	https://github.com/apache/seatunnel/commit/d7cd601051	2.3.1
[Improve][Project] Code format with spotless plugin. (#4101)	https://github.com/apache/seatunnel/commit/a2ab166561	2.3.1
[Feature][Connector] add get source method to all source connector (#3846)	https://github.com/apache/seatunnel/commit/417178fb84	2.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)	https://github.com/apache/seatunnel/commit/226dc6a119	2.3.0
[Improve][Connector-V2][Redis] Unified exception for redis source & sink exception (#3517)	https://github.com/apache/seatunnel/commit/205f782585	2.3.0
options in conditional need add to required or optional options (#3501)	https://github.com/apache/seatunnel/commit/51d5bcba10	2.3.0
[feature][api] add option validation for the ReadonlyConfig (#3417)	https://github.com/apache/seatunnel/commit/4f824fea36	2.3.0
[Feature][Redis Connector V2] Add Redis Connector Option Rules & Improve Redis Connector doc (#3320)	https://github.com/apache/seatunnel/commit/1c10aacb30	2.3.0
[Connector-V2][ElasticSearch] Add ElasticSearch Source/Sink Factory (#3325)	https://github.com/apache/seatunnel/commit/38254e3f26	2.3.0
[Improve][Connector-V2][Redis] Support redis cluster connection & user authentication (#3188)	https://github.com/apache/seatunnel/commit/c7275a49cc	2.3.0
[DEV][Api] Replace SeaTunnelContext with JobContext and remove singleton pattern (#2706)	https://github.com/apache/seatunnel/commit/cbf82f755c	2.2.0-beta
[Feature][Connector-V2] Add redis sink connector (#2647)	https://github.com/apache/seatunnel/commit/71a9e4b019	2.2.0-beta
[#2606]Dependency management split (#2630)	https://github.com/apache/seatunnel/commit/fc047be69b	2.2.0-beta
[Feature][Connector-V2] Add redis source connector (#2569)	https://github.com/apache/seatunnel/commit/405f7d6f99	2.2.0-beta