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

背景情報

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

前提条件

• Tair (Redis OSS 互換) インスタンスを作成します。詳細については、「ステップ 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 -- 必須。
) WITH (
  'connector' = 'redis',
  'host' = '<yourHost>',
  'mode' = 'STRING'  -- 結果テーブルに必須。
);

WITH パラメーター

Redis 結果テーブルのデータ形式

型マッピング

例

• 結果テーブル

  • STRING データの書き込み：この例では、redis_sink 結果テーブルの user_id 列の値がキーとして使用され、login_time 列の値が値として使用されます。両方とも 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 ブローカーアドレス
      '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 列の値がキーとして使用されます。product_name、quantity、および amount 列の値は、それぞれ Redis の `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 列の値がキーとして使用されます。product_name 列の値はフィールドとして、quantity 列の値は値として使用されます。すべてのデータは 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 ブローカーアドレス
      '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,                   -- 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 列の値がキーです。user_name、email、および register_time 列の値は、それぞれ `user_name`、`email`、および `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,     -- 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;