Redis Sink

简述

Easystream 支持输出到 Redis。

示例

- source
CREATE TABLE user_log (
    user_id VARCHAR,
    item_id VARCHAR,
    category_id VARCHAR,
    behavior VARCHAR,
    ts VARCHAR
) WITH (
    'connector.type' = 'kafka',
    'connector.version' = 'universal',
    'connector.topic' = '***',
    'connector.startup-mode' = 'latest-offset',
    'connector.properties.zookeeper.connect' = 'xxx',
    'connector.properties.bootstrap.servers' = 'xxx',
    'connector.properties.group.id' = 'xxx',
    'update-mode' = 'append',
    'format.type' = 'json',
    'format.derive-schema' = 'true'
);
-- sink
CREATE TABLE pvuvage_sink (
    user_id VARCHAR,
    item_id VARCHAR
) WITH (
    'connector.type' = 'redis',
    'primary.key' = 'user_id',
    'host' = '***',
    'port' = '6379',
    'mode' = 'single',
    'db.index' = '1',
    'password' = '***',
    'connector.command' = 'LPUSH'
);
INSERT INTO pvuvage_sink
select user_id, item_id from user_log;

• Flink 维表 Join 官方文档

With 参数