Redis Sink

一、介绍

redis sink

二、支持版本

主流版本

三、插件名称

SQL	redis-x

四、参数说明

1、SQL

• connector

  • 描述：redis-x
  • 必选：是
  • 参数类型：string
  • 默认值：无
    

• url

  • 描述：localhost:6379
  • 必选：是
  • 参数类型：string
  • 默认值：无
    

• table-name

  • 描述：tableName
  • 必选：是
  • 参数类型：string
  • 默认值：无
    

• password

  • 描述：password
  • 必选：否
  • 参数类型：string
  • 默认值：无
    

• redis-type

  • 描述： redis模式（1 单机，2 哨兵， 3 集群），默认：1
  • 必选：否
  • 参数类型：string
  • 默认值：1
    

• master-name

  • 描述： 主节点名称（哨兵模式下为必填项）
  • 必选：否
  • 参数类型：string
  • 默认值：无
    

• database

  • 描述： redis 的数据库地址，默认：0
  • 必选：否
  • 参数类型：string
  • 默认值：0
    

• timeout

  • 描述：连接超时时间，默认：10000毫秒
  • 必选：否
  • 参数类型：string
  • 默认值：10000
    

• max.total

  • 描述：最大连接数 ，默认：8
  • 必选：否
  • 参数类型：string
  • 默认值：8
    

• max.idle

  • 描述：最大空闲连接数，默认：8
  • 必选：否
  • 参数类型：string
  • 默认值：8
    

• min.idle

  • 描述： 最小空闲连接数 ，默认：0
  • 必选：否
  • 参数类型：string
  • 默认值：0
    

• key.expired-time

  • 描述：redis sink的key的过期时间。默认是0（永不过期），单位是s。默认：0
  • 必选：否
  • 参数类型：string
  • 默认值：0
    

• sink.parallelism

  • 描述：sink并行度
  • 必选：否
  • 参数类型：string
  • 默认值：无

• type和mode
  • 描述：type 表示 value 的类型，mode 表示在选定的数据类型下的写入模式。
  • 选项：string/list/set/zset/hash
  • 必选：是
  • 默认值：无

type	描述	mode	说明	注意
hash	哈希	hset	向 hash 有序集合中存储这个数据，如果已经存在则覆盖	当 value 类型是 hash 时，数据源的每一行记录需要遵循相应的规范，即每一行记录除 key 以外，只能有一对 attribute 和 value，并且 attribute 必须在 value 前面，Rediswriter 方能解析出哪一个 column 是 attribute，哪一个 column 是 value。

五、数据类型

支持	BOOLEAN、TINYINT、SMALLINT、INT、BIGINT、FLOAT、DOUBLE、DECIMAL、STRING、VARCHAR、CHAR、TIMESTAMP、DATE、BINARY
暂不支持	ARRAY、MAP、STRUCT、UNION

六、脚本示例

CREATE TABLE source
(
    id          INT
    , name      STRING
    , money     decimal
    , dateone   timestamp
    , age       bigint
    , datethree timestamp
    , datesix   timestamp(6)
    , datenigth timestamp(9)
    , dtdate    date
    , dttime    time

    , afloat    float
    , adouble   double
    , aboolean  BOOLEAN
    , abigint   BIGINT
    , atinyint  TINYINT
    , avarchar  varchar
    , asmallint SMALLINT
) WITH (
      'connector' = 'stream-x'
      ,'number-of-rows' = '10'
      );

CREATE TABLE sink
(
    id          INT
    , name      STRING
    , money     decimal
    , dateone   timestamp
    , age       bigint
    , datethree timestamp
    , datesix   timestamp(6)
    , datenigth timestamp(9)
    , dtdate    date
    , dttime    time

    , afloat    float
    , adouble   double
    , aboolean  BOOLEAN
    , abigint   BIGINT
    , atinyint  TINYINT
    , avarchar  varchar
    , asmallint SMALLINT
    , primary key (id) NOT ENFORCED -- redis必须要填写，存入redis的结构是hash结构，key=tableName_primaryKey1_primaryKey2
) WITH (
      'connector' = 'redis-x' --必填
      ,'url' = 'localhost:6379' --必填，格式ip:port[,ip:port]
      ,'table-name' = 'cx' --必填
      ,'password' = '123456' -- 密码 无默认，非必填项
      ,'redis-type' = '1' -- redis模式（1 单机，2 哨兵， 3 集群），默认：1
      ,'master-name' = 'lala' -- 主节点名称（哨兵模式下为必填项）
      ,'database' = '0' -- redis 的数据库地址，默认：0
      ,'timeout' = '10000' -- 连接超时时间，默认：10000毫秒
      ,'max.total' = '5' -- 最大连接数 ，默认：8
      ,'max.idle' = '5' -- 最大空闲连接数，默认：8
      ,'min.idle' = '0' -- 最小空闲连接数 ，默认：0
      -- ,'keyExpiredTime' = '1000' -- redis sink的key的过期时间。默认是0（永不过期），单位是s。默认：0
      ,'sink.parallelism' = '3' -- sink并行度
      ,'type' = 'hash'
      ,'mode' = 'hset'
      );


INSERT INTO sink
SELECT id
     , max(name)      as name
     , max(money)     as money
     , max(dateone)   as dateone
     , max(age)       as age
     , max(datethree) as datethree
     , max(datesix)   as datesix
     , max(datenigth) as datenigth
     , max(dtdate)    as dtdate
     , max(dttime)    as dttime

     , max(afloat)    as afloat
     , max(adouble)   as adouble
     , max(aboolean)  as aboolean
     , max(abigint)   as abigint
     , max(atinyint)  as atinyint
     , max(avarchar)  as avarchar
     , max(asmallint) as asmallint
from source
group by id;