Redis CDC Pipeline Sink Connector 用于将火山引擎流式计算 Flink 版 Pipeline 作业捕获的全量数据和增量变更实时写入 Redis，支持 String 和 Hash 两种数据布局。

功能概述 #

版本兼容性 #

特色功能 #

• 多种部署模式：支持 Single Node、Cluster 和 Sentinel。
• 两种数据布局：可将整行写为 Redis String，也可按字段写入 Redis Hash。
• CDC 事件同步：支持更新、删除和主键变化，并能处理 Redis 无 Schema 场景下的字段变化。
• 自动 Key 生成：根据表标识和主键字段构造稳定的 Redis Key。
• 记录 TTL：可为每条记录设置过期时间。

前提条件 #

1. 已准备可访问的 Redis 实例，并取得相应部署模式的地址、端口和认证信息。
2. Flink 资源池与所有 Redis 节点网络可达。
3. 源表已定义稳定、唯一且非空的主键。
4. 已使用 Flink 加密变量保存 Redis 用户名和密码。
5. 已评估预计 Key 数量、Value 大小、TTL、Redis 内存容量和淘汰策略。

使用限制 #

• Connector 仅支持作为 Flink CDC Pipeline Sink。
• 每张源表必须定义主键，Connector 使用主键字段构造 Redis Key。
• Pipeline 仅支持 string、hash，不支持 list、set、zset。
• Redis Cluster 不支持选择 db；db 仅对 Single Node 和 Sentinel 生效。
• sink.ignore-delete: true 会保留源端已删除记录和主键变化前的旧 Key。
• Connector 使用 At-Least-Once 语义，故障恢复时可能重放 Redis 命令。应使用稳定 Key 和幂等写入，使最终结果收敛。
• DROP_TABLE、TRUNCATE_TABLE 事件会被忽略，不会删除或清空 Redis 数据。
• 修改 key.prefix、key.separator、路由规则或主键字段会改变 Key，可能产生新旧数据并存。
• Hash 模式的列删除和重命名不会主动扫描全部历史 Key，旧字段可能长期残留。
• Redis 是内存数据库。全量同步、长 TTL 或无 TTL 会显著增加内存使用量。

语法结构 #

source:
  type: <source_type>
  # Source 参数

sink:
  type: redis
  name: Redis Sink
  redis-mode: <single-node|cluster|sentinel>
  host: <redis_host>
  port: 6379
  username: ${secret_values.redis-username}
  password: ${secret_values.redis-password}
  value-type: hash
  key.separator: ":"
  sink.record.ttl: 1h

pipeline:
  name: <pipeline_name>
  parallelism: 2
  local-time-zone: Asia/Shanghai

参数说明 #

部署模式与连接参数 #

连接池参数 #

Key 与 Value 参数 #

TTL 与写入参数 #

类型映射 #

Redis 没有固定 Schema。String 模式将记录组装为 JSON 或单个原始字符串；Hash 模式将每个字段序列化为字符串。

使用示例 #

将 MySQL 表同步到 Redis Hash #

以下示例将 MySQL 表同步到 Redis Hash。运行前请替换地址、表名和加密变量名称。

source:
  type: mysql
  name: MySQL Source
  hostname: <mysql_host>
  port: 3306
  username: ${secret_values.mysql-username}
  password: ${secret_values.mysql-password}
  tables: <source_database>.<source_table>
  server-id: 5401-5404

sink:
  type: redis
  name: Redis Sink
  redis-mode: single-node
  host: <redis_host>
  port: 6379
  username: ${secret_values.redis-username}
  password: ${secret_values.redis-password}
  db: 0
  value-type: hash
  key.prefix: "cdc:"
  key.separator: ":"
  value.format.skip-key: true
  sink.record.ttl: 1h
  sink.ignore-delete: false
  sink.ignore-null: false
  sink.buffer-flush.max-rows: 50
  sink.buffer-flush.interval: 2s

pipeline:
  name: MySQL to Redis Pipeline
  parallelism: 2
  local-time-zone: Asia/Shanghai

预期结果。 每条源记录根据表标识和主键生成一个以 cdc: 开头的 Redis Key，非主键字段写入 Hash。源端更新会更新相应 Hash 字段，默认删除会移除整个 Key；配置 TTL 后，Key 按指定时间过期。

Cluster 与 Sentinel 连接配置 #

Cluster 模式使用以下连接参数，且不要配置 db：

redis-mode: cluster
cluster.nodes: "<redis_host_1>:6379;<redis_host_2>:6379"
username: ${secret_values.redis-username}
password: ${secret_values.redis-password}

Sentinel 模式使用以下连接参数：

redis-mode: sentinel
master.name: <redis_master_name>
sentinels.info: "<sentinel_host_1>:26379;<sentinel_host_2>:26379"
db: 0
username: ${secret_values.redis-username}
password: ${secret_values.redis-password}

Cluster 和 Sentinel 均使用 username、password 完成 Redis 认证，不配置 Sentinel 单独鉴权参数。

运行机制 #

Key 与 Value #

默认 Key 格式为 {key.prefix}{namespace}.{table}{key.separator}{pk1}{key.separator}{pk2}...。可以使用 Pipeline route 修改 namespace.table 部分。
value-type: string 时，多字段记录通常序列化为 JSON；仅有一个非主键字段且 value.format.skip-key: true 时，直接写入该字段的字符串值。value-type: hash 时，每个字段写入一个 Hash field。

数据事件 #

Schema 事件 #

• CREATE TABLE、ADD COLUMN、RENAME COLUMN、ALTER TYPE 不触发 Redis DDL，只更新 Sink 的 Schema 缓存。
• DROP COLUMN 会记录被删除字段；后续对相应 Key 执行 Hash 写入时附带 HDEL。从未再次更新的旧 Key 不会自动清理。
• DROP TABLE、TRUNCATE TABLE 不受支持，Connector 会忽略这些事件，不删除或清空 Redis Key。

TTL #

配置 sink.record.ttl 后，String 模式使用 SET key value EX ttl 原子写入，Hash 模式在写入后执行 EXPIRE。未配置时 Key 永不过期。

Checkpoint 与故障恢复 #

Connector 提供 At-Least-Once 语义。故障恢复时可能重放尚未确认完成的 Redis 命令；稳定 Key 配合 SET、HSET、HDEL、DEL 等幂等操作，可使最终结果收敛。参数级重试不等同于 Exactly-Once。

验证与排障 #

验证方法 #

1. 提交作业并确认进入运行状态，日志中没有连接、认证或主键错误。
2. 插入一条测试记录，核对 Redis Key、Value 类型和字段内容。
3. 更新普通字段，验证 String 覆盖或 Hash 字段更新。
4. 将 Hash 字段改为 null，验证默认执行 HDEL；再开启 sink.ignore-null 对比行为。
5. 修改主键，验证旧 Key 删除和新 Key 写入。
6. 删除源记录，验证默认移除整个 Key。
7. 配置短 TTL，验证 String 和 Hash 的过期时间及更新后行为。
8. 等待成功 Checkpoint 后执行受控重启，检查最终 Key 和 Value。

常见问题 #

注意事项 #

• sink.ignore-delete: true 会形成 append-only 视图，源端删除和主键变化不会清理旧 Key。
• Key 分隔符不是转义机制。主键值可能包含分隔符时，应验证不同主键组合不会生成相同 Key。
• 修改路由、Key 前缀或分隔符会改变 Key 空间，变更前应制定旧 Key 清理方案。
• Hash 字段删除和重命名不会立即清理全部历史数据。
• DROP_TABLE、TRUNCATE_TABLE 会被忽略，源端删除或清空表不会同步清理 Redis 数据。
• TTL 会影响 Redis 内存占用和数据可见时间；更新或故障重放可能改变过期时间。
• 批量写入和提高并行度会增加 Redis QPS 与连接数，使用前应核对实例规格与限流。
• 不要在 YAML 中填写明文密码，应使用 Flink 加密变量。

相关文档 #

• 火山引擎流式计算 Flink 版：Redis Connector
• 火山引擎流式计算 Flink 版：CDC Connector 列表