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

Informasi latar belakang

Alibaba Cloud Tair adalah layanan database yang kompatibel dengan protokol open source Redis. Layanan ini menyediakan penyimpanan hibrida yang memanfaatkan memori dan disk secara bersamaan. Tair memiliki arsitektur aktif-aktif dengan ketersediaan tinggi serta arsitektur kluster yang dapat diskalakan untuk memenuhi kebutuhan bisnis akan throughput tinggi, latensi rendah, dan skalabilitas fleksibel. Untuk informasi selengkapnya, lihat Apa itu Tair (Kompatibel dengan Redis OSS)?.

Prasyarat

• Buat instans Tair (Kompatibel dengan Redis OSS). Untuk informasi selengkapnya, lihat Langkah 1: Buat instans.

• Atur daftar putih. Untuk informasi selengkapnya, lihat Langkah 2: Atur daftar putih.

Batasan

• Konektor Redis hanya menyediakan semantik best-effort dan tidak dapat menjamin pengiriman exactly-once. Oleh karena itu, Anda harus memastikan idempotensi.

• Batasan berikut berlaku untuk tabel dimensi:

  • Anda hanya dapat membaca data tipe STRING dan HASHMAP dari penyimpanan data Redis.

  • Semua bidang dalam tabel dimensi harus bertipe STRING. Anda harus mendeklarasikan tepat satu primary key.

  • Saat melakukan join dengan tabel dimensi, klausa ON harus mencakup kondisi ekuivalen untuk primary key.

Masalah yang diketahui dan solusi

Ververica Runtime (VVR) 8.0.9 memiliki masalah yang diketahui terkait fitur caching-nya. Untuk mengatasi masalah ini, Anda dapat menonaktifkan fitur tersebut dengan mengatur 'sink.buffer-flush.max-rows' = '0' di klausa WITH tabel sink.

Sintaksis

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

Parameter WITH

Format data untuk tabel sink Redis

Pemetaan tipe

Contoh

• Tabel sink

  • Menulis data STRING: Pada contoh ini, nilai kolom user_id pada tabel sink redis_sink digunakan sebagai kunci, sedangkan nilai kolom login_time digunakan sebagai nilai, lalu keduanya ditulis ke Redis.

    CREATE TEMPORARY TABLE kafka_source (
      user_id STRING,          -- User ID
      login_time STRING        -- Logon timestamp
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'user_logins',      -- Kafka topic
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker address
      'format' = 'json',            -- The data format is JSON.
      'scan.startup.mode' = 'earliest-offset' -- Start consuming from the 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',   -- Use STRING mode.
      'host' = 'yourHost',
      'port' = 'yourPort',
      'password' = 'yourPassword'
    );
    
    INSERT INTO redis_sink
    SELECT *
    FROM kafka_source;

  • Menulis data HASHMAP dalam mode multi-nilai: Pada contoh ini, nilai kolom order_id pada tabel sink redis_sink digunakan sebagai kunci. Nilai kolom product_name, quantity, dan amount masing-masing ditulis ke field `product_name`, `quantity`, dan `amount` di Redis.

    CREATE TEMPORARY TABLE kafka_source (
      order_id STRING,     -- Order ID
      product_name STRING, -- Product name
      quantity STRING,         -- Product quantity
      amount STRING         -- Order amount
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'orders_topic', -- Kafka topic
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker address
      'format' = 'json',        -- The data format is JSON.
      'scan.startup.mode' = 'earliest-offset' -- Start consuming from the earliest offset.
    );
    
    CREATE TEMPORARY TABLE redis_sink (
      order_id STRING,        -- Order ID, used as the Redis key.
      product_name STRING,    -- Product name, used as a Redis Hash field.
      quantity STRING,        -- Product quantity, used as a Redis Hash field.
      amount STRING,          -- Order amount, used as a Redis Hash field.
      PRIMARY KEY (order_id) NOT ENFORCED
    ) WITH (
      'connector' = 'redis',
      'mode' = 'HASHMAP',   -- Use HASHMAP mode.
      'flattenHash' = 'true',
      'host' = 'yourHost',
      'port' = 'yourPort',
      'password' = 'yourPassword'
    );
    
    INSERT INTO redis_sink
    SELECT *
    FROM kafka_source;

  • Menulis data HASHMAP dalam mode single-value: Pada contoh ini, nilai kolom order_id pada tabel sink redis_sink digunakan sebagai kunci. Nilai kolom product_name digunakan sebagai field, sedangkan nilai kolom quantity digunakan sebagai nilai, lalu semua data tersebut ditulis ke Redis.

    CREATE TEMPORARY TABLE kafka_source (
      order_id STRING,          -- Order ID
      product_name STRING,      -- Product name
      quantity STRING           -- Product quantity
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'orders_topic', -- Kafka topic
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker address
      'format' = 'json',        -- The data format is JSON.
      'scan.startup.mode' = 'earliest-offset' -- Start consuming from the 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;

• Tabel dimensi

  • Membaca data STRING: Pada contoh ini, nilai kolom user_id pada tabel dimensi redis_dim berfungsi sebagai kunci, sedangkan nilai kolom user_name berfungsi sebagai nilai.

    CREATE TEMPORARY TABLE kafka_source (
      user_id STRING,               -- User ID
      proctime AS PROCTIME()        -- Processing time
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'user_clicks',      -- Kafka topic
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker address
      'format' = 'json',            -- The data format is JSON.
      'scan.startup.mode' = 'earliest-offset' -- Start consuming from the earliest offset.
    );
    
    CREATE TEMPORARY TABLE redis_dim (
      user_id STRING,               -- User ID (Redis key)
      user_name STRING,             -- Username (Redis value)
      PRIMARY KEY (user_id) NOT ENFORCED
    ) WITH (
      'connector' = 'redis',
      'host' = 'yourHost',    -- Redis host address
      'port' = 'yourPort',    -- Redis port
      'password' = 'yourPassword',  -- Redis password
      'mode' = 'STRING'             -- Use STRING mode.
    );
    
    CREATE TEMPORARY TABLE blackhole_sink (
      user_id STRING,               -- User ID
      redis_user_id STRING,         -- User ID in Redis
      user_name STRING              -- Username
    ) WITH (
      'connector' = 'blackhole'
    );
    
    INSERT INTO blackhole_sink
    SELECT 
      t1.user_id,                   -- User ID (from Kafka)
      t2.user_id,                   -- User ID in Redis
      t2.user_name                  -- Username (from 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;

  • Membaca data HASHMAP dalam mode multi-nilai: Pada contoh ini, nilai kolom user_id pada tabel dimensi redis_dim berfungsi sebagai kunci. Nilai kolom user_name, email, dan register_time masing-masing berfungsi sebagai nilai dari field `user_name`, `email`, dan `register_time`.

    CREATE TEMPORARY TABLE kafka_source (
      user_id STRING,               -- User ID
      click_time TIMESTAMP(3),      -- Click time
      proctime AS PROCTIME()        -- Processing time
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'user_clicks',      -- Kafka topic
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker address
      'format' = 'json',            -- The data format is JSON.
      'scan.startup.mode' = 'earliest-offset'
    );
    
    CREATE TEMPORARY TABLE redis_dim (
      user_id STRING,     -- User ID (Redis key)
      user_name STRING,    -- Username (part of a Redis field-value pair)
      email STRING,        -- Email (part of a Redis field-value pair)
      register_time STRING, -- Registration time (part of a Redis field-value pair)
      PRIMARY KEY (user_id) NOT ENFORCED
    ) WITH (
      'connector' = 'redis',
      'host' = 'yourHost',
      'port' = 'yourPort',
      'password' = 'yourPassword',
      'mode' = 'HASHMAP'    -- Use HASHMAP mode.
    );
    
    CREATE TEMPORARY TABLE blackhole_sink (
      user_id STRING,        -- User ID
      user_name STRING,      -- Username
      email STRING,          -- Email
      register_time STRING,   -- Registration time
      click_time TIMESTAMP(3) -- Click time
    ) WITH (
      'connector' = 'blackhole'
    );
    
    INSERT INTO blackhole_sink
    SELECT 
      t1.user_id,    -- User ID
      t2.user_name,  -- Username
      t2.email,      -- Email
      t2.register_time,  -- Registration time
      t1.click_time      -- 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 single-value: Pada contoh ini, nilai `testKey` dari parameter hashName berfungsi sebagai kunci. Nilai kolom user_id pada tabel dimensi redis_dim berfungsi sebagai field, sedangkan nilai kolom user_name berfungsi sebagai nilai.

    CREATE TEMPORARY TABLE kafka_source (
      user_id STRING,            -- User ID
      proctime AS PROCTIME()     -- Processing time
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'user_clicks',      -- Kafka topic
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker address
      'format' = 'json',            -- The data format is JSON.
      'scan.startup.mode' = 'earliest-offset' -- Start consuming from the earliest offset.
    );
    
    CREATE TEMPORARY TABLE redis_dim (
      user_id STRING,               -- User ID (Redis hash field)
      user_name STRING,             -- Username (Redis hash value)
      PRIMARY KEY (user_id) NOT ENFORCED
    ) WITH (
      'connector' = 'redis',
      'host' = 'yourHost',        -- Redis host address
      'port' = 'yourPort',        -- Redis port
      'password' = 'yourPassword',-- Redis password
      'hashName' = 'testkey'      -- Fixed Redis hash name
    );
    
    CREATE TEMPORARY TABLE blackhole_sink (
      user_id STRING,               -- User ID
      redis_user_id STRING,         -- User ID in Redis
      user_name STRING              -- Username
    ) WITH (
      'connector' = 'blackhole'
    );
    
    INSERT INTO blackhole_sink
    SELECT 
       t1.user_id,     -- User ID (from Kafka)
       t2.user_id,     -- User ID in Redis
       t2.user_name    -- Username (from 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;