云数据库Tair（Redis开源版）

背景信息

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

前提条件

• 已创建云数据库Tair（Redis开源版）实例，详情请参见步骤1：创建实例。

• 已设置白名单，详情请参见步骤2：设置白名单。

使用限制

• 目前Redis连接器是仅提供Best Effort语义，无法保证数据的Exactly Once，需要您自行保证语义上的幂等性。

• 维表使用限制有：

  • 仅支持读取Redis数据存储中STRING和HASHMAP类型的数据。

  • 维表的字段必须为STRING，且必须声明且只能声明一个主键。

  • 维表JOIN时，ON条件必须包含主键的等值条件。

已知缺陷及解决方案

实时计算引擎VVR 8.0.9版本缓存功能存在问题，需要在结果表WITH参数中添加 'sink.buffer-flush.max-rows' = '0' 禁用。

语法结构

CREATE TABLE redis_table (
  col1 STRING,
  col2 STRING,
  PRIMARY KEY (col1) NOT ENFORCED -- 必填。
) WITH (
  'connector' = 'redis',
  'host' = '<yourHost>',
  'mode' = 'STRING'  -- 结果表时必填。
);

WITH参数

Redis结果表数据类型格式

类型映射

使用示例

• 结果表

  • 写入STRING类型数据：在代码示例中，redis_sink结果表中user_id列的值会作为key，login_time列的值会作为value写入到Redis中。

    CREATE TEMPORARY TABLE kafka_source (
      user_id STRING,          -- 用户ID
      login_time STRING        -- 登录时间戳
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'user_logins',      -- Kafka主题
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker地址
      'format' = 'json',            -- 数据格式为JSON
      'scan.startup.mode' = 'earliest-offset' -- 从最早的消息开始消费
    );
    
    CREATE TEMPORARY TABLE redis_sink (
      user_id STRING,               -- Redis的key
      login_time STRING,            -- Redis的value
      PRIMARY KEY (user_id) NOT ENFORCED
    ) WITH (
      'connector' = 'redis',
      'mode' = 'STRING',   -- 使用STRING模式
      'host' = 'yourHost',
      'port' = 'yourPort',
      'password' = 'yourPassword'
    );
    
    INSERT INTO redis_sink
    SELECT *
    FROM kafka_source;

  • 多值模式写入HASHMAP类型数据：在代码示例中，redis_sink结果表中的order_id列的值会作为key，product_name列的值会作为field为product_name的value，quantity列的值会作为field为quantity的value，amount列的值会作为field为amount的value，写入到Redis中。

    CREATE TEMPORARY TABLE kafka_source (
      order_id STRING,     -- 订单 ID
      product_name STRING, -- 商品名称
      quantity STRING,         -- 商品数量
      amount STRING         -- 订单金额
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'orders_topic', -- Kafka主题
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker地址
      'format' = 'json',        -- 数据格式为JSON
      'scan.startup.mode' = 'earliest-offset' -- 从最早的消息开始消费
    );
    
    CREATE TEMPORARY TABLE redis_sink (
      order_id STRING,        -- 订单ID，作为Redis的key
      product_name STRING,    -- 商品名称，作为Redis Hash的field
      quantity STRING,        -- 商品数量，作为Redis Hash的field
      amount STRING,          -- 订单金额，作为Redis Hash的field
      PRIMARY KEY (order_id) NOT ENFORCED
    ) WITH (
      'connector' = 'redis',
      'mode' = 'HASHMAP',   -- 使用HASHMAP模式
      'flattenHash' = 'true',
      'host' = 'yourHost',
      'port' = 'yourPort',
      'password' = 'yourPassword'
    );
    
    INSERT INTO redis_sink
    SELECT *
    FROM kafka_source;

  • 单值模式写入HASHMAP类型数据：在代码示例中，redis_sink结果表中的order_id列的值会作为key，product_name列的值会作为field，quantity列的值会作为value写入到Redis中。

    CREATE TEMPORARY TABLE kafka_source (
      order_id STRING,          -- 订单ID
      product_name STRING,      -- 商品名称
      quantity STRING           -- 商品数量
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'orders_topic', -- Kafka主题
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker地址
      'format' = 'json',        -- 数据格式为JSON
      'scan.startup.mode' = 'earliest-offset' -- 从最早的消息开始消费
    );
    
    CREATE TEMPORARY TABLE redis_sink (
      order_id STRING,          -- Redis的key
      product_name STRING,      -- Redis的field
      quantity STRING,          -- Redis的value
      PRIMARY KEY (order_id) NOT ENFORCED
    ) WITH (
      'connector' = 'redis',
      'mode' = 'HASHMAP',
      'host' = 'yourHost',
      'port' = 'yourPort',
      'password' = 'yourPassword'
    );
    
    INSERT INTO redis_sink
    SELECT *
    FROM kafka_source;

• 维表

  • 读取STRING类型数据：在代码示例中，redis_dim维表中的user_id列的值对应key，user_name列的值对应value。

    CREATE TEMPORARY TABLE kafka_source (
      user_id STRING,               -- 用户ID
      proctime AS PROCTIME()        -- 处理时间
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'user_clicks',      -- Kafka主题
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker地址
      'format' = 'json',            -- 数据格式为JSON
      'scan.startup.mode' = 'earliest-offset' -- 从最早的消息开始消费
    );
    
    CREATE TEMPORARY TABLE redis_dim (
      user_id STRING,               -- 用户ID（Redis key）
      user_name STRING,             -- 用户名（Redis value）
      PRIMARY KEY (user_id) NOT ENFORCED
    ) WITH (
      'connector' = 'redis',
      'host' = 'yourHost',    -- Redis主机地址
      'port' = 'yourPort',    -- Redis端口
      'password' = 'yourPassword',  -- Redis密码
      'mode' = 'STRING'             -- 使用STRING模式
    );
    
    CREATE TEMPORARY TABLE blackhole_sink (
      user_id STRING,               -- 用户ID
      redis_user_id STRING,         -- Redis中的用户ID
      user_name STRING              -- 用户名
    ) WITH (
      'connector' = 'blackhole'
    );
    
    INSERT INTO blackhole_sink
    SELECT 
      t1.user_id,                   -- 用户ID（来自 Kafka）
      t2.user_id,                   -- Redis中的用户ID
      t2.user_name                  -- 用户名（来自Redis）
    FROM kafka_source AS t1
    JOIN redis_dim FOR SYSTEM_TIME AS OF t1.proctime AS t2
    ON t1.user_id = t2.user_id;

  • 多值模式读取HASHMAP类型数据：在代码示例中，redis_dim维表中的user_id列的值对应key，user_name列的值对应field为user_name的value，email列的值对应field为email的value，register_time列的值对应field为register_time的value。

    CREATE TEMPORARY TABLE kafka_source (
      user_id STRING,               -- 用户ID
      click_time TIMESTAMP(3),      -- 点击时间
      proctime AS PROCTIME()        -- 处理时间
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'user_clicks',      -- Kafka主题
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker地址
      'format' = 'json',            -- 数据格式为JSON
      'scan.startup.mode' = 'earliest-offset'
    );
    
    CREATE TEMPORARY TABLE redis_dim (
      user_id STRING,     -- 用户ID（Redis key）
      user_name STRING,    -- 用户名（Redis field-value对的一部分）
      email STRING,        -- 邮箱（Redis field-value对的一部分）
      register_time STRING, -- 注册时间（Redis field-value对的一部分）
      PRIMARY KEY (user_id) NOT ENFORCED
    ) WITH (
      'connector' = 'redis',
      'host' = 'yourHost',
      'port' = 'yourPort',
      'password' = 'yourPassword',
      'mode' = 'HASHMAP'    -- 使用HASHMAP模式
    );
    
    CREATE TEMPORARY TABLE blackhole_sink (
      user_id STRING,        -- 用户ID
      user_name STRING,      -- 用户名
      email STRING,          -- 邮箱
      register_time STRING,   -- 注册时间
      click_time TIMESTAMP(3) -- 点击时间
    ) WITH (
      'connector' = 'blackhole'
    );
    
    INSERT INTO blackhole_sink
    SELECT 
      t1.user_id,    -- 用户ID
      t2.user_name,  -- 用户名
      t2.email,      -- 邮箱
      t2.register_time,  -- 注册时间
      t1.click_time      -- 点击时间
    FROM kafka_source AS t1
    JOIN redis_dim FOR SYSTEM_TIME AS OF t1.proctime AS t2
    ON t1.user_id = t2.user_id;

  • 单值模式读取HASHMAP类型数据：在代码示例中，hashName参数的值testKey为key，redis_dim维表中的user_id列的值对应field，user_name列的值对应value。

    CREATE TEMPORARY TABLE kafka_source (
      user_id STRING,            -- 用户ID
      proctime AS PROCTIME()     -- 处理时间
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'user_clicks',      -- Kafka主题
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker地址
      'format' = 'json',            -- 数据格式为JSON
      'scan.startup.mode' = 'earliest-offset' -- 从最早的消息开始消费
    );
    
    CREATE TEMPORARY TABLE redis_dim (
      user_id STRING,               -- 用户ID（Redis hash field）
      user_name STRING,             -- 用户名（Redis hash value）
      PRIMARY KEY (user_id) NOT ENFORCED
    ) WITH (
      'connector' = 'redis',
      'host' = 'yourHost',        -- Redis主机地址
      'port' = 'yourPort',        -- Redis端口
      'password' = 'yourPassword',-- Redis密码
      'hashName' = 'testKey'      -- 固定的Redis hash名称
    );
    
    CREATE TEMPORARY TABLE blackhole_sink (
      user_id STRING,               -- 用户ID
      redis_user_id STRING,         -- Redis中的用户ID
      user_name STRING              -- 用户名
    ) WITH (
      'connector' = 'blackhole'
    );
    
    INSERT INTO blackhole_sink
    SELECT 
       t1.user_id,     -- 用户ID（来自Kafka）
       t2.user_id,     -- Redis中的用户ID
       t2.user_name    -- 用户名（来自 Redis）
    FROM kafka_source AS t1
    JOIN redis_dim FOR SYSTEM_TIME AS OF t1.proctime AS t2
    ON t1.user_id = t2.user_id;