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

Catatan penggunaan opsi cache

• ALL memerlukan VVR 8.0.3 atau versi lebih baru.

• Untuk VVR 8.0.3 hingga VVR 11.1 (tidak termasuk), cache = ALL hanya membaca HASHMAP dalam mode nilai tunggal. Dalam klausa WITH DDL, atur hashName ke nama kunci; deklarasikan Field sebagai kunci primer dan Value sebagai kolom non-kunci primer.

• Mulai VVR 11.1, cache = ALL mendukung mode multi-nilai HASHMAP. Tentukan kunci Redis sebagai kunci primer dan deklarasikan beberapa kolom non-kunci primer untuk setiap bidang Hash. Atur mode = HASHMAP dalam klausa WITH.

• Opsi cache harus digunakan bersama dengan cacheSize dan cacheTTLMs.

Menulis data STRING

Contoh ini menggunakan user_id sebagai kunci Redis dan login_time sebagai nilai Redis.

CREATE TEMPORARY TABLE kafka_source (
  user_id    STRING,   -- ID Pengguna
  login_time STRING    -- Waktu login (Unix timestamp)
) 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,   -- Kunci Redis
  login_time STRING,   -- Nilai 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;

Menulis data HASHMAP dalam mode multi-nilai

Contoh ini menggunakan order_id sebagai kunci Redis dan menulis product_name, quantity, dan amount sebagai bidang Hash terpisah.

CREATE TEMPORARY TABLE kafka_source (
  order_id     STRING,  -- ID Pesanan
  product_name STRING,  -- Nama produk
  quantity     STRING,  -- Jumlah produk
  amount       STRING   -- Jumlah pesanan
) 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,  -- Kunci Redis
  product_name STRING,  -- Bidang Hash: product_name
  quantity     STRING,  -- Bidang Hash: quantity
  amount       STRING,  -- Bidang Hash: 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;

Menulis data HASHMAP dalam mode nilai tunggal

Contoh ini menggunakan order_id sebagai kunci Redis, product_name sebagai bidang Hash, dan quantity sebagai nilai Hash.

CREATE TEMPORARY TABLE kafka_source (
  order_id     STRING,  -- ID Pesanan
  product_name STRING,  -- Nama produk
  quantity     STRING   -- Jumlah produk
) 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,  -- Kunci Redis
  product_name STRING,  -- Bidang Hash Redis
  quantity     STRING,  -- Nilai Hash 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;

Membaca data STRING

Contoh ini menggunakan user_id sebagai kunci Redis dan mengambil user_name sebagai nilai 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,   -- Kunci Redis
  user_name STRING,   -- Nilai 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;

Membaca data HASHMAP dalam mode multi-nilai

Contoh ini menggunakan user_id sebagai kunci Redis dan mengambil beberapa bidang Hash — user_name, email, dan register_time — dalam satu kali lookup.

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,   -- Kunci Redis
  user_name     STRING,   -- Bidang Hash: user_name
  email         STRING,   -- Bidang Hash: email
  register_time STRING,   -- Bidang Hash: 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;

Membaca data HASHMAP dalam mode nilai tunggal

Contoh ini menggunakan kunci Hash tetap (testkey) yang diatur melalui hashName. Kolom user_id dipetakan ke bidang Hash, dan user_name dipetakan ke nilai Hash.

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,   -- Bidang Hash
  user_name STRING,   -- Nilai Hash
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'redis',
  'host'      = '<yourHost>',
  'port'      = '<yourPort>',
  'password'  = '<yourPassword>',
  'hashName'  = 'testkey'   -- Kunci Hash tetap
);

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;