Realtime Compute for Apache Flink:Tair (Redis OSS-compatible)

背景情報

Alibaba Cloud Tair は、オープンソース Redis プロトコルと互換性のあるデータベースサービスであり、ハイブリッドメモリ・ディスクストレージを提供します。Tair は、高スループット、低レイテンシー、柔軟なスケールアップ/スケールダウンといったビジネス要件を満たすために、高可用性 (HA) アクティブ/アクティブアーキテクチャとスケーラブルなクラスタアーキテクチャを使用しています。詳細については、「Tair (Redis OSS-compatible) とは？」をご参照ください。

前提条件

• Tair (Redis OSS-compatible) インスタンスを作成済みであること。詳細については、「ステップ 1: インスタンスの作成」をご参照ください。

• ホワイトリストを構成済みであること。詳細については、「ステップ 2: ホワイトリストの構成」をご参照ください。

制限事項

• Redis コネクタはベストエフォートセマンティクスを提供し、Exactly-once 配信を保証しません。ご利用のアプリケーションでべき等性を確保する必要があります。

• ディメンションテーブルには、以下の制限が適用されます。

  • Redis からは STRING および HASHMAP データの型のみを読み取ることができます。

  • ディメンションテーブル内のすべてのフィールドは STRING タイプである必要があります。プライマリキーを正確に 1 つ宣言する必要があります。

  • ディメンションテーブルを結合する場合、ON 句にはプライマリキーの同等条件を含める必要があります。

既知の問題とソリューション

Ververica Runtime (VVR) 8.0.9 のキャッシュ機能には問題があります。これを無効にするには、結果テーブルの WITH 句に sink.buffer-flush.max-rows = '0' を追加します。

構文

CREATE TABLE redis_table (
  col1 STRING,
  col2 STRING,
  PRIMARY KEY (col1) NOT ENFORCED -- Required.
) WITH (
  'connector' = 'redis',
  'host' = '<yourHost>',
  'mode' = 'STRING'  -- Required for sink tables.
);

WITH パラメーター

Redis 結果テーブルのデータ型フォーマット

タイプマッピング

使用例

• シンクテーブル

  • STRING データの書き込み: コード例では、redis_sink 結果テーブルの user_id カラムの値が Redis にキーとして書き込まれ、login_time カラムの値が値として書き込まれます。

    CREATE TEMPORARY TABLE kafka_source (
      user_id STRING,          -- ユーザー ID
      login_time STRING        -- ログオンタイムスタンプ
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'user_logins',      -- Kafka トピック
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka ブローカーアドレス
      'format' = 'json',            -- データフォーマットは JSON
      'scan.startup.mode' = 'earliest-offset' -- 最も古いオフセットから消費する
    );
    
    CREATE TEMPORARY TABLE redis_sink (
      user_id STRING,               -- Redis キー
      login_time STRING,            -- Redis 値
      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 カラムの値が Redis にキーとして書き込まれます。product_name、quantity、および amount カラムの値は、それぞれ product_name、quantity、および amount フィールドに書き込まれます。

    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 ブローカーアドレス
      'format' = 'json',        -- データフォーマットは JSON
      'scan.startup.mode' = 'earliest-offset' -- 最も古いオフセットから消費する
    );
    
    CREATE TEMPORARY TABLE redis_sink (
      order_id STRING,        -- 注文 ID (Redis キーとして使用)
      product_name STRING,    -- 製品名 (Redis Hash フィールドとして使用)
      quantity STRING,        -- 数量 (Redis Hash フィールドとして使用)
      amount STRING,          -- 注文金額 (Redis Hash フィールドとして使用)
      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 カラムの値が Redis にキーとして書き込まれます。product_name カラムの値はフィールドとして書き込まれ、quantity カラムの値は値として書き込まれます。

    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 ブローカーアドレス
      'format' = 'json',        -- データフォーマットは JSON
      'scan.startup.mode' = 'earliest-offset' -- 最も古いオフセットから消費する
    );
    
    CREATE TEMPORARY TABLE redis_sink (
      order_id STRING,          -- Redis キー
      product_name STRING,      -- Redis フィールド
      quantity STRING,          -- Redis 値
      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 カラムの値がキーに対応し、user_name カラムの値が値に対応します。

    CREATE TEMPORARY TABLE kafka_source (
      user_id STRING,               -- ユーザー ID
      proctime AS PROCTIME()        -- 処理時間
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'user_clicks',      -- Kafka トピック
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka ブローカーアドレス
      'format' = 'json',            -- データフォーマットは JSON
      'scan.startup.mode' = 'earliest-offset' -- 最も古いオフセットから消費する
    );
    
    CREATE TEMPORARY TABLE redis_dim (
      user_id STRING,               -- ユーザー ID (Redis キー)
      user_name STRING,             -- ユーザー名 (Redis 値)
      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,                   -- ユーザー ID (Redis から)
      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 カラムの値がキーです。user_name カラムの値は user_name フィールドにマッピングされます。email カラムの値は email フィールドにマッピングされます。register_time カラムの値は register_time フィールドにマッピングされます。

    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 ブローカーアドレス
      'format' = 'json',            -- データフォーマットは JSON
      'scan.startup.mode' = 'earliest-offset'
    );
    
    CREATE TEMPORARY TABLE redis_dim (
      user_id STRING,     -- ユーザー ID (Redis キー)
      user_name STRING,    -- ユーザー名 (Redis フィールドと値のペアの一部)
      email STRING,        -- メールアドレス (Redis フィールドと値のペアの一部)
      register_time STRING, -- 登録時間 (Redis フィールドと値のペアの一部)
      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 であり、キーとして使用されます。redis_dim ディメンションテーブルの user_id カラムの値はフィールドであり、user_name カラムの値は値です。

    CREATE TEMPORARY TABLE kafka_source (
      user_id STRING,            -- ユーザー ID
      proctime AS PROCTIME()     -- 処理時間
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'user_clicks',      -- Kafka トピック
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka ブローカーアドレス
      'format' = 'json',            -- データフォーマットは JSON
      'scan.startup.mode' = 'earliest-offset' -- 最も古いオフセットから消費する
    );
    
    CREATE TEMPORARY TABLE redis_dim (
      user_id STRING,               -- ユーザー ID (Redis ハッシュフィールド)
      user_name STRING,             -- ユーザー名 (Redis ハッシュ値)
      PRIMARY KEY (user_id) NOT ENFORCED
    ) WITH (
      'connector' = 'redis',
      'host' = 'yourHost',        -- Redis ホストアドレス
      'port' = 'yourPort',        -- Redis ポート
      'password' = 'yourPassword',-- Redis パスワード
      'hashName' = 'testkey'      -- 固定 Redis ハッシュ名
    );
    
    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,     -- ユーザー ID (Redis から)
       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;