Skip to main content
Version: Next

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​

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.

DatasourceDependency
RedisDownload

Options​

NameTypeRequiredDefaultDescription
hoststringYes when mode = SINGLE-Redis host for single-node mode.
portintNo6379Redis port for single-node mode.
nodeslistYes when mode = CLUSTER-Redis Cluster nodes, for example ["redis-0:6379", "redis-1:6379"].
modestringNoSINGLERedis deployment mode. Supported values are SINGLE and CLUSTER.
userstringNo-Redis ACL username.
authstringNo-Redis authentication password.
db_numintNo0Redis database index. This option is used in single-node mode.
keystringYes-Redis key, upstream field name, or key template when support_custom_key = true.
data_typestringYes-Redis data type. Supported values are KEY, STRING, HASH, LIST, SET, and ZSET.
formatstringNoJSONSerialization format used when no value field is configured. Supported values are JSON and TEXT.
field_delimiterstringNo,Field delimiter used when format = TEXT.
batch_sizeintNo10Maximum number of rows buffered before one batch write.
expirelongNo-1Key expiration time in seconds. Values less than or equal to 0 mean no expiration is set.
support_custom_keybooleanNofalseWhether to replace placeholders in key with upstream field values.
value_fieldstringNo-Upstream field used as the Redis value for KEY/STRING, LIST, SET, and ZSET.
hash_key_fieldstringNo-Upstream field used as the Redis hash field when data_type = HASH.
hash_value_fieldstringNo-Upstream field used as the Redis hash value when data_type = HASH.
multi_table_sink_replicaintNo1Writer replica count for multi-table writes.
common-optionsconfigNo-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}".

Schema Evolution​

Redis Sink supports schema evolution with SeaTunnel Zeta. When the upstream is a CDC source, enable schema-changes.enabled = true in the source configuration so schema change events are sent to the sink.

Redis is schema-less, so schema evolution does not execute DDL in Redis. Instead, when Redis Sink serializes the whole upstream row as JSON or TEXT, it refreshes the serializer after a supported schema change event. Newly added fields are included, and dropped fields are no longer written. See Schema Evolution for the supported event types.

Schema evolution does not rewrite field names configured in key, custom key placeholders, value_field, hash_key_field, or hash_value_field. Do not rename or drop a field referenced by these options while the job is running. If a configured field no longer exists, Redis Sink applies the missing-field behavior described in Write Rules, which can turn the configured field name into a literal key or value.

Before applying a schema change, Redis Sink flushes rows buffered with the previous schema. It also stores the latest schema in checkpoint state and restores that schema after recovery. Restoring a job from a checkpoint taken after a DDL while increasing the Redis sink parallelism is not currently supported.

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
ChangeCommitVersion
[Improve][Connector-V2] Use key_field_name option when reading Redis hash data (#9642)https://github.com/apache/seatunnel/commit/5d214a73052.3.12
[Feature][Redis] Add redis key into the result record (#9574)https://github.com/apache/seatunnel/commit/6e8b7c5da52.3.12
[Fix][Connector-Redis] Redis did not write successfully, but the task did not fail (#9055)https://github.com/apache/seatunnel/commit/07510ed9372.3.11
[hotfix][redis] fix npe cause by null host parameter (#8881)https://github.com/apache/seatunnel/commit/7bd58651652.3.10
[Improve][Redis] Optimized Redis connection params (#8841)https://github.com/apache/seatunnel/commit/e56f06cdf02.3.10
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[improve] update Redis connector config option (#8631)https://github.com/apache/seatunnel/commit/f1c313eea62.3.10
[Feature][Redis] Flush data when the time reaches checkpoint.interval and update test case (#8308)https://github.com/apache/seatunnel/commit/e15757bcd72.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/fcb29382862.3.9
[Feature][Redis] Flush data when the time reaches checkpoint.interval (#8198)https://github.com/apache/seatunnel/commit/2e24941e6a2.3.9
[Hotfix] Fix redis sink NPE (#8171)https://github.com/apache/seatunnel/commit/6b9074e7692.3.9
[Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
[Feature][Connector-Redis] Redis connector support delete data (#7994)https://github.com/apache/seatunnel/commit/02a35c39792.3.9
[Improve][Connector-V2] Redis support custom key and value (#7888)https://github.com/apache/seatunnel/commit/ef2c3c72832.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[improve][Redis]Redis scan command supports versions 5, 6, 7 (#7666)https://github.com/apache/seatunnel/commit/6e70cbe3342.3.8
[Improve][Connector] Add multi-table sink option check (#7360)https://github.com/apache/seatunnel/commit/2489f6446b2.3.7
[Feature][Core] Support using upstream table placeholders in sink options and auto replacement (#7131)https://github.com/apache/seatunnel/commit/c4ca74122c2.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/be37f05c072.3.6
[Feature][Kafka] Support multi-table source read (#5992)https://github.com/apache/seatunnel/commit/60104602d12.3.6
[Improve][Connector-V2]Support multi-table sink feature for redis (#6314)https://github.com/apache/seatunnel/commit/fed89ae3fc2.3.5
[Feature][Core] Upgrade flink source translation (#5100)https://github.com/apache/seatunnel/commit/5aabb14a942.3.4
[Feature][Connector-V2] Support TableSourceFactory/TableSinkFactory on redis (#5901)https://github.com/apache/seatunnel/commit/e84dcb8c102.3.4
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)https://github.com/apache/seatunnel/commit/8de74081002.3.4
[Improve][Connector-v2][Redis] Redis support select db (#5570)https://github.com/apache/seatunnel/commit/77fbbbd0ee2.3.4
Support config column/primaryKey/constraintKey in schema (#5564)https://github.com/apache/seatunnel/commit/eac76b4e502.3.4
[Feature][Connector-v2][RedisSink]Support redis to set expiration time. (#4975)https://github.com/apache/seatunnel/commit/b5321ff1d22.3.3
Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
[Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
[improve][api] Refactoring schema parse (#4157)https://github.com/apache/seatunnel/commit/b2f573a13e2.3.1
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Improve][Connector-V2][Redis] Unified exception for redis source & sink exception (#3517)https://github.com/apache/seatunnel/commit/205f7825852.3.0
options in conditional need add to required or optional options (#3501)https://github.com/apache/seatunnel/commit/51d5bcba102.3.0
[feature][api] add option validation for the ReadonlyConfig (#3417)https://github.com/apache/seatunnel/commit/4f824fea362.3.0
[Feature][Redis Connector V2] Add Redis Connector Option Rules & Improve Redis Connector doc (#3320)https://github.com/apache/seatunnel/commit/1c10aacb302.3.0
[Connector-V2][ElasticSearch] Add ElasticSearch Source/Sink Factory (#3325)https://github.com/apache/seatunnel/commit/38254e3f262.3.0
[Improve][Connector-V2][Redis] Support redis cluster connection & user authentication (#3188)https://github.com/apache/seatunnel/commit/c7275a49cc2.3.0
[DEV][Api] Replace SeaTunnelContext with JobContext and remove singleton pattern (#2706)https://github.com/apache/seatunnel/commit/cbf82f755c2.2.0-beta
[Feature][Connector-V2] Add redis sink connector (#2647)https://github.com/apache/seatunnel/commit/71a9e4b0192.2.0-beta
[#2606]Dependency management split (#2630)https://github.com/apache/seatunnel/commit/fc047be69b2.2.0-beta
[Feature][Connector-V2] Add redis source connector (#2569)https://github.com/apache/seatunnel/commit/405f7d6f992.2.0-beta