Redis数据源

DataWorks数据集成支持使用Redis Writer将数据写至Redis,本文为您介绍DataWorks的Redis数据离线写入能力。

使用限制

  • 数据导入过程支持使用Serverless资源组(推荐)独享数据集成资源组

  • 使用Redis Writer向Redis写入数据时,如果Value类型是List,重跑同步任务的同步结果不是幂等的。因此,如果Value类型是List ,重跑同步任务时,需要您手动清空Redis上相应的数据。

支持的字段类型

Redis支持较丰富的存储value类型,包括String(字符串)、List(链表)、Set(集合)、ZSet(sorted set有序集合)和Hash(哈希类型)。Redis详情请参见redis.io

数据同步任务开发

数据同步任务的配置入口和通用配置流程可参见下文的配置指导。

附录:脚本Demo与参数说明

离线任务脚本配置方式

如果您配置离线任务时使用脚本模式的方式进行配置,您需要按照统一的脚本格式要求,在任务脚本中编写相应的参数,详情请参见通过脚本模式配置离线同步任务,以下为您介绍脚本模式下数据源的参数配置详情。

Writer脚本Demo

以下以读取MySQL数据并写入Redis为例,为您示例读取端MySQL Reader和写入端Redis Writer的脚本代码样例。写入Redis的数据同步作业。

{
    "type":"job",
    "version":"2.0",  //版本号
    "steps":[
        { //以下为读取端代码样例,读取端的参数详情可查看对应数据源的Reader插件文档。
            "stepType":"mysql",   
            "parameter": {
                "envType": 0,
                "datasource": "xc_mysql_demo2",
                "column": [
                    "id",
                    "value",
                    "table"
                ],
                "connection": [
                    {
                        "datasource": "xc_mysql_demo2",
                        "table": []
                    }
                ],
                "where": "",
                "splitPk": "",
                "encoding": "UTF-8"
            },,
            "name":"Reader",
            "category":"reader"
        },
        {//以下为写入端代码样例。
            "stepType":"redis",                    //Redis Writer的插件名,配置为redis。
            "parameter":{                          //以下为Redis Writer的主要参数。
                "expireTime":{                     //Redis value值缓存失效时间,可配置为seconds类型或unixtime类型。"seconds":"1000"
                            }, 
                "keyFieldDelimiter":"u0001",       //写入Redis的key的分隔符。
                "dateFormat":"yyyy-MM-dd HH:mm:ss",//写入Redis时,Date的时间格式。
                "datasource":"xc_mysql_demo2",     //数据源名称,需与添加的数据源名称保持一致。
                "envType": 0,                      //环境类型,开发环境:1,生产环境:0。
                "writeMode":{                      //写入模式。
                    "type":"string"                //value类型。
                    "mode":"set",                  //value是某类型时,写入的模式。
                    "valueFieldDelimiter":"u0001", //value之间的分隔符。
                             },
                "keyIndexes":[0,1],                //用于源端到Redis的映射,指定源端需要作为key的列(第1列从0开始),如果源端第1列、第2列组合作为Redis的key,这里配置为[0,1]。
                "batchSize":"1000"                 //一次性批量提交的记录数大小。
        "column": [                        // 对于redis类型为string,set操作,如果此column没有配置那么value的格式是分隔符连接的字符串(csv格式,假设ID的值为1,name的值为"小王",age的值为18,sex的值为男,redis的value结果示例:"18::男");如果配置了column,且按照如下格式配置,则redis 的value将把原列的列名和值写入成JSON格式,假设ID的值为1,name的值为"小王",age的值为18,sex的值为男,redis的value结果示例{"id":1,"name":"小王","age":18,"sex":"男"}
                {
                "name": "id",
                "index": "0"

                },
                {
                "name": "name",
                "index": "1"
                },
                {
                "name": "age",
                "index": "2"
                },
                {
                "name": "sex",
                "index": "3"
                }
            ]
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"                           //错误记录数。
        },
        "speed":{
            "throttle":true,//当throttle值为false时,mbps参数不生效,表示不限流;当throttle值为true时,表示限流。
            "concurrent":1, //作业并发数。
            "mbps":"12"//限流,此处1mbps = 1MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Writer脚本参数

参数

描述

是否必选

默认值

expireTime

Redis value值缓存失效时间,单位为秒。如果不填该配置项,则该配置项取值为默认值0,表示永久有效。

expireTime的配置方式有以下两种:

  • seconds方式:指定了从现在开始多长时间后数据失效,取值是失效时间相对当前时间的秒数。

  • unixtime方式:指定了从1970.1.1开始多长时间后数据失效。取值是失效时间相对1970.1.1时间的秒数。

0(0表示永久有效)

keyFieldDelimiter

写入Redis的Key分隔符。例如key=key1\u0001id,如果有多个Key需要拼接时,该值为必填项。如果只有一个Key,则可以忽略该配置项。

\u0001

dateFormat

写入Redis时,Date的时间格式为yyyy-MM-dd HH:mm:ss

datasource

数据源名称。该配置项填写的内容必须与添加的数据源名称保持一致。

selectDatabase

写入Redis的库选择("0"~"databases"),Redis集群无法做数据库选择。

默认写到0号库

writeMode

Redis Writer写入Redis的value类型包含以下5种:

  • 字符串(string)

  • 字符串列表(list)

  • 字符串集合(set)

  • 有序字符串集合(zset)

  • 哈希(hash)

不同的value类型,writeMode配置会略有差异,详细说明可参见下文writeMode参数说明

说明

配置Redis Writer时,您需要配置writeMode为支持的5种写入数据类型中的1种类型,且只能配置1种。如果您没有配置,则writeMode取值为默认值string

string

keyIndexes

指定作为key的源端列的列序号。列序号从0开始(即第1列的序号是0,第2列的序号是1,依次类推)。

  • 源端的某一列作为Redis的key时:配置为对应列的序号即可,例如,第1列作为key,则配置为0

  • 源端的连续多列组合作为Redis的key时:配置为对应多列的序号数组,例如,第2列至第4列组合作为key,则配置为[1,3]

说明

配置keyIndexes后,Redis Writer会将其余的列作为Value。如果您只想同步源表的某几列作为Key,某几列作为Value,则无需同步所有字段,在Reader插件端指定好column进行列筛选即可。

batchSize

一次性批量提交的记录数大小。该值可以极大减少数据同步系统与Redis的网络交互次数,并提升整体吞吐量。如果该值设置过大,会导致数据同步运行进程OOM异常。

1,000

timeout

写入Redis的超时时间,单位为毫秒。

30,000

redisMode

Redis的运行模式。具体如下:

  • 集群模式:redisMode取值为ClusterMode,表示集群模式。

    集群模式下,其他数据源写入Redis时,与Redis集群进行直连通信。通常,自建Redis集群地址阿里云Redis直连地址需使用此模式。集群模式暂不支持批量写入。

  • 非集群模式:redisMode取值为空(即不配置任何值),表示非集群模式。

    通常,阿里云Redis集群代理地址、读写分离地址、标准版地址使用此模式,支持批量写入。

说明

支持使用Serverless资源组(推荐)和独享数据集成资源组。

column

写入Redis的column配置。对于Redis对应类型为string,set操作时:

  • 如果此column没有配置,那么value的格式是分隔符连接的字符串。(csv格式,假设ID的值为1,name的值为"小王",age的值为18,sex的值为男,Redis的value结果示例:"18::男");

  • 如果配置了column,且按照如下格式配置,比如"column": [{"index":"0", "name":"id"}, {"index":"1", "name":"name"}] , 这样Redis的value数据写出到Redis后,以 {"id":"对应源头列的值","name":"对应源头列的值"} 的JSON形式存储,假设ID的值为1,name的值为"小王",Redis的value结果示例{"id":1,"name":"小王"}”

writeMode参数说明

配置Redis Writer时,您需要配置writeMode为支持的5种写入数据类型中的1种类型,且只能配置1种。如果您没有配置,则writeMode取值为默认值string

value类型

type参数(必选)

mode参数(必选)

valueFieldDelimiter参数(非必选)

writeMode配置样例

字符串(string)

type需配置为string

mode为写入模式参数,value为字符串(string)时:

  • mode需配置为set

  • 如果需存储的数据已经存在,则覆盖原有的数据。

valueFieldDelimiter为value之间的分隔符,默认值为\u0001

  • 该配置项主要用于源数据每行超过两列的情况,例如有三列时,各列通过分隔符分割样例为value1\u0001value2\u0001value3

  • 如果源数据只有两列(即key和value)时,则无需配置。

"writeMode":{
        "type": "string",
        "mode": "set",
        "valueFieldDelimiter": "\u0001"
        }

字符串列表(list)

type需配置为list

mode为写入模式参数,value为字符串列表(list)时,可配置为:

  • lpush,表示在list最左边存储数据。

  • rpush,表示在list最右边存储数据。

"writeMode":{
    "type": "list",
    "mode": "lpush|rpush",
    "valueFieldDelimiter": "\u0001"
}

字符串集合(set)

type需配置为set

mode为写入模式参数,value为字符串集合(set)时:

  • mode需配置为sadd,表示向set集合中存储数据。

  • 如果需存储的数据已经存在,则覆盖原有的数据。

"writeMode":{
        "type": "set",
        "mode": "sadd",
        "valueFieldDelimiter": "\u0001"
        }

有序字符串集合(zset)

type需配置为zset

mode为写入模式参数,value为有序字符串集合(zset)时:

  • mode需配置为zadd,表示向zset有序集合中存储数据。

  • 如果需存储的数据已经存在,则覆盖原有的数据。

无需配置此参数。

"writeMode":{
        "type": "zset",
        "mode": "zadd"
        }
说明

当value类型为zset时,数据源的每行记录均需遵循相应的规范。即每行记录除key外,只能有1对score和value,并且score必须在value前面,Redis Writer方可解析出column对应的是score或value。

哈希(hash)

type需配置为hash

mode为写入模式参数,value为哈希(hash)时:

  • mode需配置为hset,表示向hash有序集合中存储数据。

  • 如果需存储的数据已经存在,则覆盖原有的数据。

无需配置此参数。

"writeMode":{
        "type": "hash",
        "mode": "hset"
        }
说明

当value类型为hash时,数据源的每行记录都需遵循相应的规范。即每行记录除key外,只能有1对attribute和value,并且attribute必须在value前面,Redis Writer方可解析出column对应的是attribute或value。