Realtime Compute for Apache Flink:Tair (Redis OSS-compatible) コネクタ

cache オプションの注意事項

• ALL は VVR 8.0.3 以降が必要です。

• VVR 8.0.3 から VVR 11.1 (除く) まで、cache = ALL は HASHMAP を単一値モードでのみ読み取ります。DDL の WITH 句で、hashName をキー名に設定し、Field をプライマリキーとして、Value を非プライマリキー列として宣言します。

• VVR 11.1 以降、cache = ALL は HASHMAP の複数値モードをサポートします。Redis キーをプライマリキーとして指定し、各ハッシュフィールドに対して複数の非プライマリキー列を宣言します。WITH 句で mode = HASHMAP を設定します。

• cache オプションは cacheSize および cacheTTLMs と一緒に使用する必要があります。

STRING データの書き込み

この例では、user_id を Redis キーとして、login_time を Redis の値として使用します。

CREATE TEMPORARY TABLE kafka_source (
  user_id    STRING,   -- ユーザー ID
  login_time STRING    -- ログイン時刻 (UNIX タイムスタンプ)
) WITH (
  'connector'                     = 'kafka',
  'topic'                         = 'user_logins',
  'properties.bootstrap.servers'  = '<yourKafkaBroker>',
  'format'                        = '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',
  'host'      = '<yourHost>',
  'port'      = '<yourPort>',
  'password'  = '<yourPassword>'
);

INSERT INTO redis_sink
SELECT * FROM kafka_source;

HASHMAP データの複数値モードでの書き込み

この例では、order_id を 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',
  'properties.bootstrap.servers'  = '<yourKafkaBroker>',
  'format'                        = 'json',
  'scan.startup.mode'             = 'earliest-offset'
);

CREATE TEMPORARY TABLE redis_sink (
  order_id     STRING,  -- Redis キー
  product_name STRING,  -- ハッシュフィールド：product_name
  quantity     STRING,  -- ハッシュフィールド：quantity
  amount       STRING,  -- ハッシュフィールド：amount
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector'   = 'redis',
  'mode'        = 'HASHMAP',
  'flattenHash' = 'true',
  'host'        = '<yourHost>',
  'port'        = '<yourPort>',
  'password'    = '<yourPassword>'
);

INSERT INTO redis_sink
SELECT * FROM kafka_source;

HASHMAP データの単一値モードでの書き込み

この例では、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',
  'properties.bootstrap.servers'  = '<yourKafkaBroker>',
  'format'                        = '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 データの読み取り

この例では、user_id を Redis キーとして使用し、user_name を Redis の値として取得します。

CREATE TEMPORARY TABLE kafka_source (
  user_id  STRING,
  proctime AS PROCTIME()
) WITH (
  'connector'                     = 'kafka',
  'topic'                         = 'user_clicks',
  'properties.bootstrap.servers'  = '<yourKafkaBroker>',
  'format'                        = 'json',
  'scan.startup.mode'             = 'earliest-offset'
);

CREATE TEMPORARY TABLE redis_dim (
  user_id   STRING,   -- Redis キー
  user_name STRING,   -- Redis の値
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'redis',
  'host'      = '<yourHost>',
  'port'      = '<yourPort>',
  'password'  = '<yourPassword>',
  'mode'      = 'STRING'
);

CREATE TEMPORARY TABLE blackhole_sink (
  user_id       STRING,
  redis_user_id STRING,
  user_name     STRING
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT
  t1.user_id,
  t2.user_id,
  t2.user_name
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 データの複数値モードでの読み取り

この例では、user_id を Redis キーとして使用し、複数のハッシュフィールド — user_name、email、register_time — を 1 回のルックアップで取得します。

CREATE TEMPORARY TABLE kafka_source (
  user_id    STRING,
  click_time TIMESTAMP(3),
  proctime   AS PROCTIME()
) WITH (
  'connector'                     = 'kafka',
  'topic'                         = 'user_clicks',
  'properties.bootstrap.servers'  = '<yourKafkaBroker>',
  'format'                        = 'json',
  'scan.startup.mode'             = 'earliest-offset'
);

CREATE TEMPORARY TABLE redis_dim (
  user_id       STRING,   -- Redis キー
  user_name     STRING,   -- ハッシュフィールド：user_name
  email         STRING,   -- ハッシュフィールド：email
  register_time STRING,   -- ハッシュフィールド：register_time
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'redis',
  'host'      = '<yourHost>',
  'port'      = '<yourPort>',
  'password'  = '<yourPassword>',
  'mode'      = 'HASHMAP'
);

CREATE TEMPORARY TABLE blackhole_sink (
  user_id       STRING,
  user_name     STRING,
  email         STRING,
  register_time STRING,
  click_time    TIMESTAMP(3)
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT
  t1.user_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) を使用します。user_id 列はハッシュフィールドに、user_name はハッシュ値にマッピングされます。

CREATE TEMPORARY TABLE kafka_source (
  user_id  STRING,
  proctime AS PROCTIME()
) WITH (
  'connector'                     = 'kafka',
  'topic'                         = 'user_clicks',
  'properties.bootstrap.servers'  = '<yourKafkaBroker>',
  'format'                        = 'json',
  'scan.startup.mode'             = 'earliest-offset'
);

CREATE TEMPORARY TABLE redis_dim (
  user_id   STRING,   -- ハッシュフィールド
  user_name STRING,   -- ハッシュ値
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'redis',
  'host'      = '<yourHost>',
  'port'      = '<yourPort>',
  'password'  = '<yourPassword>',
  'hashName'  = 'testkey'   -- 固定ハッシュキー
);

CREATE TEMPORARY TABLE blackhole_sink (
  user_id       STRING,
  redis_user_id STRING,
  user_name     STRING
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT
  t1.user_id,
  t2.user_id,
  t2.user_name
FROM kafka_source AS t1
JOIN redis_dim FOR SYSTEM_TIME AS OF t1.proctime AS t2
ON t1.user_id = t2.user_id;