You need to enable JavaScript to run this app.
文档中心
大数据研发治理套件

大数据研发治理套件

复制全文
下载 pdf
数据源列表
配置 Redis 数据源
复制全文
下载 pdf
配置 Redis 数据源

Redis 数据源为您提供读取和写入 Redis 数据的双向通道能力,实现不同数据源与 Redis 之间进行数据传输。
本文为您介绍 DataSail 的 Redis 数据同步的能力支持情况。

1 支持的 Redis 版本

  • 脚本模式(DSL)离线读:
    • 支持自建 Redis 单机部署模式。
  • 可视化离线读写:
    • 支持火山引擎云 NoSQL 数据库 Redis 社区版、Redis 企业版 7.X 版本。
    • 支持自建 Redis 单机部署模式。

2 使用限制

子账号新建数据源时,需要有项目的管理员角色,方可以进行新建数据源操作。各角色对应权限说明,详见:管理成员
为确保同步任务使用的独享集成资源组具有 Redis 数据库节点的网络访问能力,您需将独享集成资源组和 Redis 数据库节点网络打通,详见网络连通解决方案
如果您需要通过 VPC 网络访问 Redis 数据库,则独享集成资源组所在 VPC 中的 IPv4 CIDR 地址,需加入到 Redis 的访问白名单中:

  1. 确认集成资源组所在的 VPC:
    Image
  2. 查看 VPC 的 IPv4 CIDR 地址:

    注意

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

    Image
  3. 将获取到的 IPv4 CIDR 地址添加进 Redis 数据库白名单中。

如果您通过公网形式访问 Redis 数据库,则需进行以下操作:

  • 独享集成资源组开通公网访问能力,操作详见开通公网
  • 并将公网 IP 地址,添加进 Redis 数据库白名单中。

3 支持的字段类型

当前 Redis 数据源支持如下字段:

字段类型

离线读(Redis Reader)

离线写(Redis Writer)

String

支持

支持

Hash

支持

支持

Set

支持

支持

Sorted Set (别名Zset)

支持

支持

List

支持

不支持

4 数据同步任务开发

4.1 数据源注册

新建数据源操作详见配置数据源,下面为您介绍用不同方式来配置 Redis 数据源信息:

  1. 火山引擎 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 参数支持
    Image

    参数

    说明

    基本配置

    *数据源类型

    Redis

    *接入方式

    火山引擎 Redis

    *数据源名称

    数据源的名称,可自行设置,仅支持中文,英文,数字,“_”,100个字符以内。

    参数配置

    *Redis 实例 ID

    下拉选择已在火山引擎 NoSQL 数据库中创建的 Redis 社区版或企业版实例名称信息。
    若您还未创建 Redis 实例,您可前往 Redis 实例控制台中创建,详见创建实例

    *用户名

    输入有权限访问数据库的用户名信息。

    注意

    输入的用户账号信息,其被授予的 Redis 角色中,至少应包含 read 和 connection 权限,这样数据源才能正常连通并访问 Redis。
    您可在 Redis 实例的账号管理中,创建相应的角色,并将其授予给用户。操作详见 Redis 创建角色
    Image

    *密码

    输入用户名对应的密码信息。

    Access Key ID

    进入火山引擎,访问控制台的密钥管理界面,复制 Accesskey ID 填写到此处。如果为子用户,请联系主账号获取密钥。

    Secret Access Key

    与 Access Key ID 配套使用,类似登录密码,用于签名您的访问参数,以防被篡改。

  2. 普通 Redis 数据源连接串方式
    其中参数名称前带 * 的为必填参数,名称前未带 * 的为可选填参数。

    参数

    说明

    基本配置

    *数据源类型

    Redis

    *接入方式

    连接串

    *数据源名称

    数据源的名称,可自行设置,仅支持中文,英文,数字,“_”,100个字符以内。

    参数配置

    *主机名或 IP 地址

    填写 Redis 数据库所在的主机名或 IP 地址信息。

    *端口

    用于访问 Redis 的端口信息,一般为6379。

    用户名

    输入有权限访问数据库的用户名信息。

    密码

    填写用于登录 Redis 的密码信息。

4.2 新建离线任务

Redis 数据源测试连通性成功后,进入到数据开发界面,开始新建 Redis 相关通道任务。
新建任务方式详见离线数据同步

4.3 可视化配置说明

任务创建成功后,您可根据实际场景,配置 Redis 离线读Redis 离线写通道任务。

4.3.1 Redis 离线读

数据来源端选择 Redis,并完成以下相关参数配置:
其中参数名称前带 * 的为必填参数,名称前未带 * 的为可选填参数。

参数

说明

*数据源类型

下拉选择Redis数据源类型。

*数据源名称

已在数据源管理中注册成功的Redis数据源,下拉可选。
若还未建立相应数据源,可单击数据源管理按钮,前往创建Redis数据源。

4.3.2 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.3.3 字段映射

字段为一一映射关系,即将源表字段信息,同步到同一行的目标字段信息中。选择好来源和目标数据源信息后,您可进行以下操作。
字段映射支持选择基础模式转换模式配置映射:

注意

基础模式和转换模式不支持互相切换,模式切换后,将清空现有字段映射中所有配置信息,一旦切换无法撤销,需谨慎操作。

  • 转换模式:
    字段映射支持数据转换,您可根据实际业务需求进行配置,将源端采集的数据,事先通过数据转换后,以指定格式输入到目标端数据库中。
    转换模式详细操作说明详见4.1 转换模式
    在转换模式中,你可依次配置:来源节点、数据转换、目标节点信息:

    配置节点

    说明

    来源节点

    配置数据来源 Source 节点信息:

    • 节点名称:自定义输入来源节点名称信息,只允许由数字、字母、下划线、-和.组成;且长度不能超过10。
    • 数据字段:通过自动添加、手动添加等方式添加数据来源字段信息。

    配置完成后,单击确认按钮,完成来源节点配置。

    数据转换

    单击数据转换右侧添加按钮,选择 SQL 转换方式,配置转换信息和规则:

    • 节点名称:自定义输入来源节点名称信息,只允许由数字、字母、下划线、-和.组成;且长度不能超过10。
    • SQL 脚本:输入 SQL 脚本转换规则,目前仅支持添加一个转换的 SQL 语句,且不能包括 “;”。

    配置完成后,单击确认按钮,完成数据转换节点配置。SQL 脚本示例详见4.1.2 添加转换节点

    目标节点

    配置目标节点 Sink 信息:

    • 节点名称:自定义输入来源节点名称信息,只允许由数字、字母、下划线、-和.组成;且长度不能超过10。
    • 数据字段:通过自动添加、手动添加等方式添加数据目标字段信息。

    配置完成后,单击确认按钮,完成目标节点配置。

  • 基础模式:

    • Redis离线读字段配置,默认有以下7个字段信息组成,暂不支持手动调整或删除:
      • key:Redis 中 key 内容数据。
      • type:支持数据格式类型 kv,set,list,map,zset。
      • expire:数据过期时间。
      • value:数据格式为 string 时,对应的数据内容。
      • values:数据格式为 set | list 时,对应的数据内容。
      • hash:数据格式为 hash 时,对应的数据内容。
      • zset:数据格式为 zset 时,对应的数据内容。
    • Redis 目标端字段映射配置,根据上面配置的写入字段类型,自动获取字段映射。

      说明

      • 基础模式不支持手动添加和删除字段;
      • 转换模式可按需写转换语句,如 Hash 格式下,可通过 Transform 语句,写入多个 Key-Value,示例转换语句:SELECT `name` AS `key`, '语文成绩', `语文成绩` FROM Source UNION SELECT `name` AS `key`, '数学成绩', `数学成绩` FROM Source UNION SELECT `name` AS `key`, '物理成绩', `物理成绩` FROM Source

4.4 DSL 配置说明

Redis 数据源支持使用脚本模式(DSL)的方式进行配置。
在某些复杂场景下,或当数据源类型暂不支持可视化配置时,您可通过任务脚本的方式,按照统一的 Json 格式,编写 Redis Reader 或 Redis Writer 参数脚本代码,来运行数据集成任务。

4.4.1 进入 DSL 模式

进入 DSL 模式操作流程,可详见 MySQL 数据源-4.4.1 进入DSL 模式

4.4.2 Redis Reader

根据实际情况替换 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,默认为:
com.bytedance.bitsail.connector.redis.source.RedisSource

*nodes

Redis 服务连接地址,填写形式为: hostname:port 或者 ip:port

auth

输入 Redis 服务安全认证密码。

*columns

Redis 读取的源表字段为固定结构,须按照文档配置,不可变更

[
    {
        "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>"
    }
]

4.4.3 Redis Writer

根据实际情况替换 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,默认为:
com.bytedance.dts.batch.redis.VolcRedisOutputFormat

ttl_type

设置 Redis Key 过期时间单位:

  • day:天级
  • hour:小时级
  • minute:分钟级

day

ttl

设置 Redis Key 过期时间,具体实现时会以 ttl * ttl_type 的形式,转换为对应秒数。

-1,永不过期

*redis_data_type

配置写入 Redis 的数据类型,支持以下四种数据类型:

  • string
  • hash
  • set
  • sorted_set

*columns

配置写入 Redis 的字段结构,写入字段映射须按照文档配置。
Redis 写入字段结构配置与 redis_data_type 字段类型对应:

  • string

    [
        {
            "upperCaseName":"KEY",
            "name":"key",
            "type":"string"
        },
        {
            "upperCaseName":"VALUE",
            "name":"value",
            "type":"string"
        }
    ]
    
  • hash

    [
        {
            "upperCaseName":"REDIS_KEY",
            "name":"redis_key",
            "type":"string"
        },
        {
            "upperCaseName":"HASH_KEY",
            "name":"hash_key",
            "type":"string"
        },
        {
            "upperCaseName":"VALUE",
            "name":"value",
            "type":"string"
        }
    ]
    
  • set

    [
        {
            "upperCaseName":"KEY",
            "name":"key",
            "type":"string"
        },
        {
            "upperCaseName":"VALUE",
            "name":"value",
            "type":"string"
        }
    ]
    
  • sorted_set

    [
        {
            "upperCaseName":"REDIS_KEY",
            "name":"redis_key",
            "type":"string"
        },
        {
            "upperCaseName":"SCORE",
            "name":"score",
            "type":"string"
        },
        {
            "upperCaseName":"VALUE",
            "name":"value",
            "type":"string"
        }
    ]
    

5 高级参数说明

  • 对于可视化通道任务,配置到自定义参数设置中,且读参数需要加上 job.reader.,写参数需要加上 job.writer. 前缀,如下图所示:
    Image
  • 对于 DSL 任务,读参数请配置到 reader.parameter 下,写参数请配置到 writer.parameter 下,直接输入参数名称和参数值。如下图所示:
    Image

5.1 Redis 离线读

参数名

描述

默认值

job.reader.key_pattens

读取 Redis keys 的匹配策略,默认为*,即读取所有的 key。
支持精确匹配和模糊匹配:

  • 精确匹配:比如只同步user_auser_b的 key,填写user_a,user_b即可;
  • 模糊匹配:比如只同步前缀是user_info的 key,填写 user_info*即可。

*

job.reader.db_index

Redis 逻辑库索引号,默认为 0。如果您的数据位于 Redis 的其他 DB,比如在 DB 6 中,则填写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 读取模式:

  • scan 模式:当读取的 key 内容数据非常多时,任务可能会出现如 JedisConnectionException: Unexpected end of stream 等异常信息,此时读取 Redis 推荐使用 scan 模式,该模式会将 key 内容分批次去读取,每批次读取一部分内容,提高同步性能。
  • all 模式:默认把 Redis 中所有的 key 内容都一次性全量拉出来,然后再处理。

all

job.reader.scan_batch_size

与 job.reader.read_mode 参数一起使用。当 job.reader.read_mode=scan 模式时,需通过该参数指定每次拉取的 size,默认每次读取 10000 个 key。

10000

5.2 Redis 离线写

参数名

描述

默认值

job.writer.write_batch_interval

一次性批量提交的数据条数,该值可以减少与 Redis 网络的交互次数并提升整体吞吐量。但如果该值设置过大可能会导致数据同步进程 OOM。

50

job.writer.database

指定写入 Redis 中的 Database 信息,默认为0。

0

附录 Redis 目的地写入配置说明

概述

DataLeap 支持将上游数据写入 Redis 的四种数据结构:stringhashsetsorted_set。不同模式对应不同的 Redis 写入命令和字段映射方式,用户可在任务配置的"目的地"模块中选择写入模式并完成字段映射。
Image

String 写入模式

  • 适用场景
    适合将单个字段值或经过转换的结果存储为 Redis 字符串类型,常见于缓存用户属性、存储序列化 JSON 等场景。

  • 写入语义

    SET <key> <value> [EX seconds]
    
  • 多字段配置示例
    如果上游数据表中有多个字段,可使用转换模式配置,如下图所示。
    Image
    转换模式需先确认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`
    

Hash 写入模式

  • 适用场景
    适合将一行数据的多个字段以 field-value 形式存储到同一个 Hash Key 下,常见于存储实体对象(如用户信息、商品详情)。

  • 写入语义

    HSET <key> <field1> <value1> [<field2> <value2> ...]
    
  • 示例

    • 上游数据表结构
      Image

    • 转换模式的 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"
      

Set 写入模式

  • 适用场景
    适合将上游字段值作为元素追加到 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"
      

Sorted_Set 写入模式

  • 适用场景
    适合需要按分值排序的集合场景,常见于排行榜、优先级队列、时间序列索引等。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 适合大多数场景。

最近更新时间:2026.04.21 14:07:26
这个页面对您有帮助吗?
有用
有用
无用
无用