版本：Next

Redis

Redis 接收器连接器

描述

Redis 接收器连接器可以在批处理或流处理作业中把上游数据写入 Redis。它支持单节点 Redis 和 Redis Cluster，可以写入 key/string、hash、list、set、zset 这几类 Redis 数据。

key 可以是固定的 Redis key，也可以是上游字段名。开启 support_custom_key = true 后，还可以用上游字段拼出 Redis key，例如 user:${id}。

支持引擎

Spark
Flink
SeaTunnel Zeta

主要特性

支持的数据源信息

使用 Redis 连接器时，需要安装以下依赖。可以通过 install-plugin.sh 安装，也可以从 Maven 中央仓库下载。

数据源	依赖
Redis	下载

选项

名称	类型	是否必填	默认值	描述
host	string	`mode = SINGLE` 时必填	-	单节点模式下的 Redis 主机地址。
port	int	否	6379	单节点模式下的 Redis 端口。
nodes	list	`mode = CLUSTER` 时必填	-	Redis Cluster 节点，例如 `["redis-0:6379", "redis-1:6379"]`。
mode	string	否	SINGLE	Redis 部署模式，支持 `SINGLE` 和 `CLUSTER`。
user	string	否	-	Redis ACL 用户名。
auth	string	否	-	Redis 认证密码。
db_num	int	否	0	Redis 数据库编号，主要用于单节点模式。
key	string	是	-	Redis key、上游字段名，或开启 `support_custom_key` 后的 key 模板。
data_type	string	是	-	Redis 数据类型，支持 `KEY`、`STRING`、`HASH`、`LIST`、`SET`、`ZSET`。
format	string	否	JSON	未配置 value 字段时的序列化格式，支持 `JSON` 和 `TEXT`。
field_delimiter	string	否	`,`	`format = TEXT` 时使用的字段分隔符。
batch_size	int	否	10	每次批量写入前最多缓存的行数。
expire	long	否	-1	Redis key 过期时间，单位秒。小于等于 0 表示不设置过期时间。
support_custom_key	boolean	否	false	是否用上游字段值替换 `key` 里的占位符。
value_field	string	否	-	写入 `KEY`/`STRING`、`LIST`、`SET`、`ZSET` 时，作为 Redis value 的上游字段。
hash_key_field	string	否	-	`data_type = HASH` 时，作为 Redis hash field 的上游字段。
hash_value_field	string	否	-	`data_type = HASH` 时，作为 Redis hash value 的上游字段。
multi_table_sink_replica	int	否	1	多表写入时的写入器副本数。
common-options	config	否	-	接收器插件通用参数，详情请参考接收器通用选项。

写入规则

key

当 support_custom_key = false 时，连接器会先判断 key 是否是上游字段名：

如果字段存在，就使用该字段的值作为 Redis key。
如果字段不存在，就把 key 本身当作固定 Redis key。

当 support_custom_key = true 时，连接器会用上游字段值替换 key 里的占位符。${field} 和旧格式 {field} 都支持。

data_type

KEY 和 STRING：每行写入一个 Redis 字符串。同一个 key 下，后写入的数据会覆盖先写入的数据。
HASH：写入 Redis hash。配置 hash_key_field 可以指定 hash field。
LIST：把每行数据追加到 Redis list。
SET：把每行数据加入 Redis set。
ZSET：把每行数据加入 Redis zset，score 固定为 1。

value_field

写入 KEY/STRING、LIST、SET、ZSET 时，如果只想写入某一个上游字段，可以配置 value_field。如果不配置，连接器会按 format 把整行数据序列化后写入 Redis。

hash_key_field 和 hash_value_field

写入 HASH 时，hash_key_field 用来指定 Redis hash field。如果配置了 hash_value_field，则把这个字段的值作为 Redis hash value；如果不配置，连接器会把整行数据序列化后作为 hash value。

multi_table_sink_replica

多表写入时的写入器副本数。当上游数据带有表标识，并且一个作业需要写入多张 Redis 表时使用。

多表作业中，key 可以包含 ${table_name}，这样不同上游表的数据会写入不同 Redis key，例如 key = "redis-result-${table_name}"。

示例

写入 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"
  }
}

使用自定义 Key 模板

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

写入 Redis Hash

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

写入 Redis Cluster 并设置过期时间

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
  }
}

变更日志

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

Redis

描述​

支持引擎​

主要特性​

支持的数据源信息​

选项​

写入规则​

key​

data_type​

value_field​

hash_key_field 和 hash_value_field​

multi_table_sink_replica​

示例​

写入 Redis List​

使用自定义 Key 模板​

写入 Redis Hash​

写入 Redis Cluster 并设置过期时间​

变更日志​

描述