本文为您介绍云数据库Redis维表DDL定义、WITH参数、CACHE参数、类型映射和代码示例。

说明 Flink支持读取自建Redis服务的数据。

什么是云数据库Redis版

阿里云数据库Redis版是兼容开源Redis协议标准、提供内存加硬盘混合存储的数据库服务,基于高可靠双机热备架构及可平滑扩展的集群架构,充分满足高吞吐、低延迟及弹性变配的业务需求。

前提条件

使用限制

  • 仅Flink计算引擎VVR 2.0.0及以上版本支持云数据库Redis Connector。
  • Redis维表仅支持读取Redis数据存储中STRING类型的数据。

DDL定义

CREATE TABLE redis_dim (
  id STRING,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'redis',
  'host' = '<yourHost>',
  'port' = '<yourPort>',
  'password' = '<yourPassword>',
  'dbNum' = '<yourDbNum>'
);
说明
  • Redis维表必须声明且只能声明一个主键。
  • Redis维表仅支持声明两个字段,且字段类型必须为STRING。
  • 维表JOIN时,ON条件必须包含所有主键的等值条件。
  • Redis仅支持None和LRU两种缓存策略。

WITH参数

参数 说明 是否必填 备注
connector 维表类型。 固定值为redis
host Redis连接地址。 无。
port Redis连接端口。 默认值为6379。
dbNum 选择操作的数据库。 默认值为0。
password Redis密码。 默认值为空,不进行权限验证。
clusterMode Redis集群是否为cluster模式。 取值如下:
  • true:为cluster模式。
  • false(默认值):不为cluster模式。
hashName Hash模式下的Hash Key名称。 默认值为空。通常,Redis维表中的数据类型为STRING类型,即key-value对。如果设置hashName参数,则Redis维表中的数据类型为HASHMAP类型,即key-{field-value}对,其中:
  • keyhashName参数值。
  • field为您在CREATE TABLE中指明的key参数值。
  • valuekey对应的赋值,和STRING类型key-valuevalue语义相同。

CACHE参数

参数 说明 是否必填 备注
cache 缓存策略。 云数据库Redis维表支持以下两种缓存策略:
  • None(默认值):无缓存。
  • LRU:缓存维表里的部分数据。源表的每条数据都会触发系统先在Cache中查找数据,如果没有找到,则去物理维表中查找。

    需要配置相关参数:缓存大小(cacheSize)和缓存更新时间间隔(cacheTTLMs)。

cacheSize 缓存大小。 选择LRU缓存策略后,可以设置缓存大小,默认为10000行。
cacheTTLMs 缓存超时时长。 默认缓存不超时,单位为毫秒。可选LRU缓存策略,即设置缓存失效的超时时长。
cacheEmpty 是否缓存空结果。 默认值为true。

类型映射

Redis字段类型 Flink字段类型
STRING VARCHAR

代码示例

CREATE TEMPORARY TABLE datagen_source (
  id STRING, 
  data STRING,
  proctime as PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE redis_dim (
  id STRING,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED --Redis中的Row Key字段。
) WITH (
  'connector' = 'redis',
  'host' = '<yourHost>',
  'port' = '<yourPort>',
  'password' = '<yourPassword>'
);

CREATE TEMPORARY blackhole_sink (
  id STRING,
  data STRING,
  name STRING
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT e.*, w.*
FROM datagen_source AS e
JOIN redis_dim FOR SYSTEM_TIME AS OF e.proctime AS w
ON e.id = w.id;