Realtime Compute for Apache Flink：雲資料庫Tair（Redis開源版）

背景資訊

阿里雲資料庫Tair是相容開源Redis協議標準、提供記憶體加硬碟混合儲存的資料庫服務，基於高可靠雙機熱備架構及可平滑擴充的叢集架構，充分滿足高吞吐、低延遲及彈性變更配置的業務需求，更多內容詳情請參見什麼是雲資料庫 Tair（相容 Redis）。

前提條件

• 已建立雲資料庫Tair（Redis開源版）執行個體，詳情請參見步驟1：建立執行個體。

• 已設定白名單，詳情請參見步驟2：設定白名單。

使用限制

• 目前Redis連接器是僅提供Best Effort語義，無法保證資料的Exactly Once，需要您自行保證語義上的等冪性。

• 維表使用限制有：

  • 僅支援讀取Redis資料存放區中STRING和HASHMAP類型的資料。

  • 維表的欄位必須為STRING，且必須聲明且只能聲明一個主鍵。

  • 維表JOIN時，ON條件必須包含主鍵的等值條件。

已知缺陷及解決方案

Realtime Compute引擎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;