Redis 数据源为您提供读取和写入 Redis 数据的双向通道能力,实现不同数据源与 Redis 之间进行数据传输。
本文为您介绍 DataSail 的 Redis 数据同步的能力支持情况。
子账号新建数据源时,需要有项目的管理员角色,方可以进行新建数据源操作。各角色对应权限说明,详见:管理成员。
为确保同步任务使用的独享集成资源组具有 Redis 数据库节点的网络访问能力,您需将独享集成资源组和 Redis 数据库节点网络打通,详见网络连通解决方案。
如果您需要通过 VPC 网络访问 Redis 数据库,则独享集成资源组所在 VPC 中的 IPv4 CIDR 地址,需加入到 Redis 的访问白名单中:

注意
若考虑安全因素,减少 IP CIDR 的访问范围,您至少需要将集成资源组绑定的子网下的 IPv4 CIDR 地址加入到数据库白名单中。

如果您通过公网形式访问 Redis 数据库,则需进行以下操作:
当前 Redis 数据源支持如下字段:
字段类型 | 离线读(Redis Reader) | 离线写(Redis Writer) |
|---|---|---|
String | 支持 | 支持 |
Hash | 支持 | 支持 |
Set | 支持 | 支持 |
Sorted Set (别名Zset) | 支持 | 支持 |
List | 支持 | 不支持 |
新建数据源操作详见配置数据源,下面为您介绍用不同方式来配置 Redis 数据源信息:
火山引擎 Redis 数据源方式
其中参数名称前带 * 的为必填参数,名称前未带 * 的为可选填参数。
说明
数据源接入方式为火山引擎 Redis,且配置 Access Key ID、Secret Access Key 参数时,此时离线读取 Redis,会调用 Redis 服务,通过 Dump 模式,来读取解析 Redis 的 RDB 文件,将数据传入目标端。
若不配置 Access Key ID、Secret Access Key 参数,或通过连接串形式访问 Redis 数据源时,在您读取 Redis 数据之前,需要先在 Redis 实例的参数配置中,解除 disabled-commands 参数对 Keys 命令默认的禁用限制,从而直接读取 Redis 数据。参数说明详见 Redis 参数支持。
参数 | 说明 |
|---|---|
基本配置 | |
*数据源类型 | Redis |
*接入方式 | 火山引擎 Redis |
*数据源名称 | 数据源的名称,可自行设置,仅支持中文,英文,数字,“_”,100个字符以内。 |
参数配置 | |
*Redis 实例 ID | 下拉选择已在火山引擎 NoSQL 数据库中创建的 Redis 社区版或企业版实例名称信息。 |
*用户名 | 输入有权限访问数据库的用户名信息。 注意 输入的用户账号信息,其被授予的 Redis 角色中,至少应包含 read 和 connection 权限,这样数据源才能正常连通并访问 Redis。 |
*密码 | 输入用户名对应的密码信息。 |
Access Key ID | 进入火山引擎,访问控制台的密钥管理界面,复制 Accesskey ID 填写到此处。如果为子用户,请联系主账号获取密钥。 |
Secret Access Key | 与 Access Key ID 配套使用,类似登录密码,用于签名您的访问参数,以防被篡改。 |
普通 Redis 数据源连接串方式
其中参数名称前带 * 的为必填参数,名称前未带 * 的为可选填参数。
参数 | 说明 |
|---|---|
基本配置 | |
*数据源类型 | Redis |
*接入方式 | 连接串 |
*数据源名称 | 数据源的名称,可自行设置,仅支持中文,英文,数字,“_”,100个字符以内。 |
参数配置 | |
*主机名或 IP 地址 | 填写 Redis 数据库所在的主机名或 IP 地址信息。 |
*端口 | 用于访问 Redis 的端口信息,一般为6379。 |
用户名 | 输入有权限访问数据库的用户名信息。 |
密码 | 填写用于登录 Redis 的密码信息。 |
Redis 数据源测试连通性成功后,进入到数据开发界面,开始新建 Redis 相关通道任务。
新建任务方式详见离线数据同步。
任务创建成功后,您可根据实际场景,配置 Redis 离线读 或 Redis 离线写通道任务。
数据来源端选择 Redis,并完成以下相关参数配置:
其中参数名称前带 * 的为必填参数,名称前未带 * 的为可选填参数。
参数 | 说明 |
|---|---|
*数据源类型 | 下拉选择Redis数据源类型。 |
*数据源名称 | 已在数据源管理中注册成功的Redis数据源,下拉可选。 |
数据目标端选择 Redis,并完成以下相关参数配置:
其中参数名称前带 * 的为必填参数,名称前未带 * 的为可选填参数。
参数 | 说明 |
|---|---|
*目标类型 | 数据去向目标类型选择 Redis。 |
*数据源名称 | 已在数据源管理界面注册的 Redis 数据源,下拉可选。 |
*数据保存时间 | 写入 Redis 数据时,为每条 key 设置过期时间,单位可以选择天或小时级,配置为 "0" 代表永久。 |
*写入字段类型 | 下拉选择写入 Redis Value 的数据类型,支持选择 string、set、hash、sorted_set 四种类型。相关配置说明请参见附录 Redis 目的地写入配置说明。 |
*写入前删除Key | 当写入字段类型选择为 set、hash、sorted_set 时,您可根据需求选择是否开启“写入前删除Key”按钮。开启该按钮后,在执行任务前,系统会先执行 Delete 语句,删除Redis中已存在的相同key。 |
字段为一一映射关系,即将源表字段信息,同步到同一行的目标字段信息中。选择好来源和目标数据源信息后,您可进行以下操作。
字段映射支持选择基础模式和转换模式配置映射:
注意
基础模式和转换模式不支持互相切换,模式切换后,将清空现有字段映射中所有配置信息,一旦切换无法撤销,需谨慎操作。
转换模式:
字段映射支持数据转换,您可根据实际业务需求进行配置,将源端采集的数据,事先通过数据转换后,以指定格式输入到目标端数据库中。
转换模式详细操作说明详见4.1 转换模式
在转换模式中,你可依次配置:来源节点、数据转换、目标节点信息:
配置节点 | 说明 |
|---|---|
来源节点 | 配置数据来源 Source 节点信息:
配置完成后,单击确认按钮,完成来源节点配置。 |
数据转换 | 单击数据转换右侧添加按钮,选择 SQL 转换方式,配置转换信息和规则:
配置完成后,单击确认按钮,完成数据转换节点配置。SQL 脚本示例详见4.1.2 添加转换节点。 |
目标节点 | 配置目标节点 Sink 信息:
配置完成后,单击确认按钮,完成目标节点配置。 |
基础模式:
说明
Redis 数据源支持使用脚本模式(DSL)的方式进行配置。
在某些复杂场景下,或当数据源类型暂不支持可视化配置时,您可通过任务脚本的方式,按照统一的 Json 格式,编写 Redis Reader 或 Redis Writer 参数脚本代码,来运行数据集成任务。
进入 DSL 模式操作流程,可详见 MySQL 数据源-4.4.1 进入DSL 模式。
根据实际情况替换 Redis Reader 相应参数,Redis Reader 脚本示例如下:
// 变量使用规则如下: // 1.自定义参数变量: {{}}, 比如{{number}} // 2.系统时间变量${}, 比如 ${date}、${hour} // ************************************** { // [required] dsl version, suggest to use latest version "version": "0.2", // [required] execution mode, supoort streaming / batch now "type": "batch", // reader config "reader": { // [required] datasource type "type": "redis", // [optional] datasource id, set it if you have registered datasource "datasource_id": 12345, // [required] user parameter "parameter": { // ********** please write here ********** // "key" : value "class":"com.bytedance.bitsail.connector.redis.source.RedisSource", "nodes":"xxx.xx.xx.xx:xx", "auth":"xxxxxxxxx", "columns": [ { "upperCaseName":"KEY", "name":"key", "type":"string" }, { "upperCaseName":"TYPE", "name":"type", "type":"string" }, { "upperCaseName":"EXPIRE", "name":"expire", "type":"bigint" }, { "upperCaseName":"VALUE", "name":"value", "type":"string" }, { "upperCaseName":"VALUES", "name":"values", "type":"array<string>" }, { "upperCaseName":"HASH", "name":"hash", "type":"MAP<string,string>" }, { "upperCaseName":"ZSET", "name":"zset", "type":"MAP<string,string>" } ] }, // writer config "writer": { }, // common config "common": { // [required] user parameter "parameter": { // ********** please write here ********** // "key" : value } } }
Reader 参数说明,其中参数名称前带 * 的为必填参数,名称前未带 * 的为可选填参数:
参数名 | 描述 | 默认值 |
|---|---|---|
*type | 数据源类型,对应 Redis 类型,默认填写为:redis | 无 |
*datasource_id | 注册的 Redis 数据源 ID,可以在项目控制台 > 数据源管理界面中查找。 | 无 |
*class | 配置 Redis 读取的 ClassName,默认为: | 无 |
*nodes | Redis 服务连接地址,填写形式为: | 无 |
auth | 输入 Redis 服务安全认证密码。 | 无 |
*columns | Redis 读取的源表字段为固定结构,须按照文档配置,不可变更
| 无 |
根据实际情况替换 Redis Writer 相应参数,Redis Writer 脚本示例如下:
// ************************************** // 变量使用规则如下: // 1.自定义参数变量: {{}}, 比如{{number}} // 2.系统时间变量${}, 比如 ${date}、${hour} // ************************************** { // [required] dsl version, suggest to use latest version "version": "0.2", // [required] execution mode, supoort streaming / batch now "type": "batch", // reader config "reader": { // [required] datasource type "type": "xx", // [optional] datasource id, set it if you have registered datasource "datasource_id": null, // [required] user parameter "parameter": { // ********** please write here ********** // "key" : value } }, // writer config "writer": { // [required] datasource type "type": "redis", // [optional] datasource id, set it if you have registered datasource "datasource_id": 12345, // [required] user parameter "parameter": { // ********** please write here ********** // "key" : value "class":"com.bytedance.dts.batch.redis.VolcRedisOutputFormat", "ttl_type": "DAY", "ttl": -1, "redis_data_type":"string", "columns": [ { "name": "name_sample", "type": "type_sample" } ] } }, // common config "common": { // [required] user parameter "parameter": { // ********** please write here ********** // "key" : value } } }
Writer 参数说明,其中参数名称前带 * 的为必填参数,名称前未带 * 的为可选填参数:
参数名 | 描述 | 默认值 |
|---|---|---|
*type | 数据源类型,对应 Redis 类型,默认填写为:redis | 无 |
*datasource_id | 注册的 Redis 数据源 ID,可以在项目控制台 > 数据源管理界面中查找。 | 无 |
*class | 配置 Redis 写入的 ClassName,默认为: | 无 |
ttl_type | 设置 Redis Key 过期时间单位:
| day |
ttl | 设置 Redis Key 过期时间,具体实现时会以 ttl * ttl_type 的形式,转换为对应秒数。 | -1,永不过期 |
*redis_data_type | 配置写入 Redis 的数据类型,支持以下四种数据类型:
| 无 |
*columns | 配置写入 Redis 的字段结构,写入字段映射须按照文档配置。
| 无 |
job.reader.,写参数需要加上 job.writer. 前缀,如下图所示:
reader.parameter 下,写参数请配置到 writer.parameter 下,直接输入参数名称和参数值。如下图所示:
参数名 | 描述 | 默认值 |
|---|---|---|
job.reader.key_pattens | 读取 Redis keys 的匹配策略,默认为
| * |
job.reader.db_index | Redis 逻辑库索引号,默认为 0。如果您的数据位于 Redis 的其他 DB,比如在 DB 6 中,则填写 | 0 |
job.reader.reader_parallelism_num | 读取 Redis 分片数,Redis 服务为单线程模型,推荐设置为 1,默认为 1。 | 1 |
job.reader.client_timeout_ms | 创建 Redis 连接的超时时间,单位为毫秒(ms)。 | 60000 |
job.reader.max_attempt_count | 执行单次 Redis Command 失败的最大重试次数 | 3 |
job.reader.read_mode | 指定 Redis 读取模式:
| all |
job.reader.scan_batch_size | 与 job.reader.read_mode 参数一起使用。当 job.reader.read_mode=scan 模式时,需通过该参数指定每次拉取的 size,默认每次读取 10000 个 key。 | 10000 |
参数名 | 描述 | 默认值 |
|---|---|---|
job.writer.write_batch_interval | 一次性批量提交的数据条数,该值可以减少与 Redis 网络的交互次数并提升整体吞吐量。但如果该值设置过大可能会导致数据同步进程 OOM。 | 50 |
job.writer.database | 指定写入 Redis 中的 Database 信息,默认为0。 | 0 |
DataLeap 支持将上游数据写入 Redis 的四种数据结构:string、hash、set、sorted_set。不同模式对应不同的 Redis 写入命令和字段映射方式,用户可在任务配置的"目的地"模块中选择写入模式并完成字段映射。
适用场景
适合将单个字段值或经过转换的结果存储为 Redis 字符串类型,常见于缓存用户属性、存储序列化 JSON 等场景。
写入语义
SET <key> <value> [EX seconds]
多字段配置示例
如果上游数据表中有多个字段,可使用转换模式配置,如下图所示。
转换模式需先确认Source 的结构,再添加数据转换的节点(支持 Flink SQL),最后确认目标节点。
在转换模式的 SQL 样例如下:
SELECT `id` AS `key`, -- 将 id 作为key,写入 redis key CONCAT_WS('|', CAST(`brand_id` AS STRING), CAST(`brand_name` AS STRING), CAST(`customer_id` AS STRING), CAST(`customer_name` AS STRING), CAST(`customer_hospital` AS STRING), CAST(`type_id` AS STRING), CAST(`type_name` AS STRING), CAST(`uv` AS STRING), CAST(`avg_stay_time` AS STRING), CAST(`mt` AS STRING), CAST(`is_deleted` AS STRING), CAST(`create_user` AS STRING), CAST(`update_user` AS STRING), CAST(`create_time` AS STRING), CAST(`update_time` AS STRING) ) AS `value` -- 将其他字段拼接成一个字符串,写入 redis value FROM `Source`
适用场景
适合将一行数据的多个字段以 field-value 形式存储到同一个 Hash Key 下,常见于存储实体对象(如用户信息、商品详情)。
写入语义
HSET <key> <field1> <value1> [<field2> <value2> ...]
示例
上游数据表结构
转换模式的 SQL 样例
SELECT * FROM ( -- brand_id SELECT CAST(`id` AS STRING) AS `key`, 'brand_id' AS `field`, CAST(`brand_id` AS STRING) AS `value` FROM `Source` UNION ALL -- brand_name SELECT CAST(`id` AS STRING) AS `key`, 'brand_name' AS `field`, CAST(`brand_name` AS STRING) AS `value` FROM `Source` UNION ALL -- customer_id SELECT CAST(`id` AS STRING) AS `key`, 'customer_id' AS `field`, CAST(`customer_id` AS STRING) AS `value` FROM `Source` UNION ALL -- customer_name SELECT CAST(`id` AS STRING) AS `key`, 'customer_name' AS `field`, CAST(`customer_name` AS STRING) AS `value` FROM `Source` UNION ALL -- customer_hospital SELECT CAST(`id` AS STRING) AS `key`, 'customer_hospital' AS `field`, CAST(`customer_hospital` AS STRING) AS `value` FROM `Source` UNION ALL -- type_id SELECT CAST(`id` AS STRING) AS `key`, 'type_id' AS `field`, CAST(`type_id` AS STRING) AS `value` FROM `Source` UNION ALL -- type_name SELECT CAST(`id` AS STRING) AS `key`, 'type_name' AS `field`, CAST(`type_name` AS STRING) AS `value` FROM `Source` UNION ALL -- uv SELECT CAST(`id` AS STRING) AS `key`, 'uv' AS `field`, CAST(`uv` AS STRING) AS `value` FROM `Source` UNION ALL -- avg_stay_time SELECT CAST(`id` AS STRING) AS `key`, 'avg_stay_time' AS `field`, CAST(`avg_stay_time` AS STRING) AS `value` FROM `Source` UNION ALL -- mt SELECT CAST(`id` AS STRING) AS `key`, 'mt' AS `field`, CAST(`mt` AS STRING) AS `value` FROM `Source` UNION ALL -- is_deleted SELECT CAST(`id` AS STRING) AS `key`, 'is_deleted' AS `field`, CAST(`is_deleted` AS STRING) AS `value` FROM `Source` UNION ALL -- create_user SELECT CAST(`id` AS STRING) AS `key`, 'create_user' AS `field`, CAST(`create_user` AS STRING) AS `value` FROM `Source` UNION ALL -- update_user SELECT CAST(`id` AS STRING) AS `key`, 'update_user' AS `field`, CAST(`update_user` AS STRING) AS `value` FROM `Source` UNION ALL -- create_time SELECT CAST(`id` AS STRING) AS `key`, 'create_time' AS `field`, CAST(`create_time` AS STRING) AS `value` FROM `Source` UNION ALL -- update_time SELECT CAST(`id` AS STRING) AS `key`, 'update_time' AS `field`, CAST(`update_time` AS STRING) AS `value` FROM `Source` ) t;
实际写入效果
HSET 1001 brand_id "5" brand_name "Nike" customer_id "88" customer_name "John" customer_hospital "PLA" type_id "3" type_name "premium" uv "120" avg_stay_time "45" mt "2024-04-01" is_deleted "0" create_user "admin" update_user "editor" create_time "2024-01-01 00:00:00" update_time "2024-04-01 12:00:00"
适用场景
适合将上游字段值作为元素追加到 Redis Set 中,自动去重,常见于标签集合、关注列表、去重 ID 池等场景。
写入语义
SADD <key> <member>
示例
转换模式的 SQL 样例
SELECT `key`, `member` FROM ( SELECT CAST(`id` AS STRING) AS `key`, CAST(`tag_1` AS STRING) AS `member` FROM `Source` UNION ALL SELECT CAST(`id` AS STRING) AS `key`, CAST(`tag_2` AS STRING) AS `member` FROM `Source` UNION ALL SELECT CAST(`id` AS STRING) AS `key`, CAST(`tag_3` AS STRING) AS `member` FROM `Source` ) t
实际写入效果
SADD 1001 "sports" SADD 1001 "music" SADD 1001 "outdoor"
适用场景
适合需要按分值排序的集合场景,常见于排行榜、优先级队列、时间序列索引等。Score 支持来自上游字段(包括经 Flink SQL 转换后的字段),也支持配置为常量。
写入语义
ZADD <key> <score> <member>
示例
转换模式的 SQL 样例
SELECT `key`, `score`, `member` FROM ( -- 按浏览时间排序的商品 SELECT CAST(`id` AS STRING) AS `key`, UNIX_TIMESTAMP(`view_time`, 'yyyy-MM-dd HH:mm:ss') AS `score`, CAST(`item_id` AS STRING) AS `member` FROM `Source` UNION ALL -- 按购买时间排序的商品(不同字段、不同 score 来源) SELECT CAST(`id` AS STRING) AS `key`, UNIX_TIMESTAMP(`buy_time`, 'yyyy-MM-dd HH:mm:ss') AS `score`, CAST(`bought_item_id` AS STRING) AS `member` FROM `Source` ) t
实际写入效果
ZADD 1001 1714012800 "item_A" ZADD 1001 1714016400 "item_B" ZADD 1001 1714020000 "item_C"
策略 | 说明 | 适用模式 |
|---|---|---|
覆盖 | Key 已存在时覆盖原有值 | string、hash |
追加 | 向已有 Key 追加成员,不清除原数据 | set、sorted_set |
Q:Key 插值字段为空时如何处理?
该条记录会被跳过并写入错误日志,不会影响其他记录的正常写入。建议在上游数据处理阶段过滤或填充空值。
Q:Hash 全量映射会覆盖已有 field 吗?HSET 语义为逐 field 写入,仅更新本次写入涉及的 field,Hash 中其他 field 不受影响。
Q:Sorted_Set 的 member 重复时如何处理?
Redis ZADD 的行为是:若 member 已存在,则更新其 Score 值,不会产生重复成员。
Q:TTL 对 Hash / Set / Sorted_Set 是否同样生效?
是,TTL 作用于整个 Key,对所有数据类型生效。写入时会对 Key 执行 EXPIRE 命令。
Q:批量提交大小如何影响性能?
Pipeline 批次越大,网络往返次数越少,吞吐量越高,但单批次失败时影响的数据量也越大。建议根据单条数据大小和 Redis 实例配置调整,默认 500 适合大多数场景。