Redis

Redis连接器支持流式写，维表join

FLINK-1.12

DDL写法

1、建表

当redis表作为sink时，需要配置connector.command

CREATE TABLE dim_table (
    user_id VARCHAR,
    age varchar
) WITH (
    'connector' = 'redis',
    'host' = '10.122.173.131',
    'port' = '6379',
    'mode' = 'single',
    'db.index' = '0',
    'password' = 'xxx',
    'lookup.cache.type' = 'all',
    'lookup.cache.ttl' = '3000',
    'primary.key' = 'user_id'
CREATE TABLE sink (
    item_id VARCHAR,
    item_type varchar
) WITH (
    'connector' = 'redis',
    'host' = '10.122.173.131',
    'port' = '6379',
    'mode' = 'single',
    'db.index' = '0',
    'password' = 'xxx',
    'lookup.cache.type' = 'all',
    'lookup.cache.ttl' = '3000',
    'primary.key' = 'item_id',
    'connector.command' = 'SET'
);

2、维表join

INSERT INTO pvuvage_sink
select d.ts as dt, d.age, count(*) as pv, count(distinct d.user_id) as uv
from (
SELECT
  u.ts,
  cast(w.age as int)  as age,
  u.user_id,
  u.behavior
FROM (select user_id,item_id,category_id,behavior,ts,PROCTIME() as proc from user_log) as u
join dim_table for system_time as of u.proc as w
on u.user_id = w.user_id
where w.age > 10
) as d
GROUP BY d.ts, d.age;

Metahub写法

1、建表

redis命令名作为属性头

set 'SADD.db.index' ='4';
set 'SADD.primary.key' = 'item_id';

2、维表join

使用metahub方式的redis作为维表时，redis命令使用GET，关联键名是k，值名是v

INSERT INTO sloth_redis_test.mem.`HSET`
select d.ts as item_id, cast(d.age as string) as item_type, cast(count(*) as string) as `redis.value`
from (
SELECT
  u.ts,
  cast(w.v as int)  as age,
  u.user_id,
  u.behavior
FROM (select user_id,item_id,category_id,behavior,ts,PROCTIME() as proc from user_log) as u
join sloth_redis_test.`default`.`GET` for system_time as of u.proc as w
on u.user_id = w.k
where w.v > 10
) as d
GROUP BY d.ts, d.age;

支持的命令

With 参数

• EasyStream Engine 使用说明
  • Release Note
    • 1.0.0
    • 1.1.0
    • 1.2.0
    • 1.2.1
    • 1.2.2
    • 1.2.3
    • 1.2.4
    • 1.3.0
    • 1.3.1
    • 1.4.0
    • 1.4.0.1
    • 1.4.1
    • 1.4.1.3
    • 1.4.2
    • 1.4.3
    • 1.4.4
    • 1.4.5
    • 1.4.6
  • Flink SQL
    • SQL 连接器
      • Hybrid SQL 连接器
      • Pulsar SQL 连接器
      • StarRocks/Doris SQL 连接器
      • Arctic SQL 连接器
      • MongoDB SQL 连接器
      • CDC SQL连接器
      • Kafka SQL 连接器
      • Upsert Kafka SQL 连接器
      • RocketMQ SQL 连接器
      • JDBC SQL 连接器
      • Elasticsearch SQL 连接器
      • HBase SQL 连接器
      • Redis SQL 连接器
      • Kudu SQL 连接器
      • Http SQL 连接器
      • TSDB SQL 连接器
      • File System SQL 连接器
      • Hive SQL 连接器
      • Print SQL 连接器
      • Black Hole SQL 连接器
      • DataGen SQL 连接器
    • Formats
      • 平台用户开发自定义 Format
      • Confluent Avro Format
      • JSON Format
      • PB Format
      • NDC Format
      • CSV Format
      • Avro Format
      • Debezium Format
      • Canal Format
      • Parquet Format
      • Orc Format
      • Raw Format
      • Bytes Format
    • Functions
      • 用户自定义函数
      • 常用函数
    • 开发
      • 混编
      • 基于元数据方式 SQL 开发
        • 维表 JOIN
          • Arctic 维表 Join
          • HBase 维表 Join
          • JDBC 维表 Join
          • Redis 维表 Join
        • Sink 表
        • Source
        • SQL 语法
      • 配置
      • SQL Hints
      • 时区
      • Schema配置
        • Metadata配置
        • 计算列
        • Schema
        • Watermark
        • 主键配置

浙公网安备 33010802008719号