Realtime Compute for Apache Flink：Tair (Kompatibel dengan Redis OSS)

Informasi latar belakang

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

Prasyarat

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

• Anda telah mengonfigurasi daftar putih. Untuk informasi selengkapnya, lihat Langkah 2: Konfigurasi daftar putih.

Batasan

• Konektor Redis menyediakan semantik best-effort dan tidak menjamin pengiriman exactly-once. Anda harus memastikan idempotensi dalam aplikasi Anda.

• Batasan berikut berlaku untuk tabel dimensi:

  • Hanya tipe data STRING dan HASHMAP yang dapat dibaca dari 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.

Isu yang diketahui dan solusi

Fitur cache pada Ververica Runtime (VVR) 8.0.9 memiliki masalah. Untuk menonaktifkannya, tambahkan sink.buffer-flush.max-rows = '0' ke klausa WITH pada 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 tipe data untuk tabel sink Redis

Pemetaan tipe

Contoh penggunaan

• Sink table

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

    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',            -- Data format is JSON
      'scan.startup.mode' = 'earliest-offset' -- Consume 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-value: Pada contoh kode, nilai kolom order_id pada tabel sink redis_sink ditulis ke Redis sebagai kunci. Nilai kolom product_name, quantity, dan amount masing-masing ditulis ke field product_name, quantity, dan amount.

    CREATE TEMPORARY TABLE kafka_source (
      order_id STRING,     -- Order ID
      product_name STRING, -- Product name
      quantity STRING,         -- Quantity
      amount STRING         -- Order amount
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'orders_topic', -- Kafka topic
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker address
      'format' = 'json',        -- Data format is JSON
      'scan.startup.mode' = 'earliest-offset' -- Consume 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,        -- 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 kode, nilai kolom order_id pada tabel sink redis_sink ditulis ke Redis sebagai kunci. Nilai kolom product_name ditulis sebagai field, sedangkan nilai kolom quantity ditulis sebagai nilai.

    CREATE TEMPORARY TABLE kafka_source (
      order_id STRING,          -- Order ID
      product_name STRING,      -- Product name
      quantity STRING           -- Quantity
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'orders_topic', -- Kafka topic
      'properties.bootstrap.servers' = 'yourKafkaBroker', -- Kafka broker address
      'format' = 'json',        -- Data format is JSON
      'scan.startup.mode' = 'earliest-offset' -- Consume 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 kode, nilai kolom user_id pada tabel dimensi redis_dim sesuai dengan kunci, sedangkan nilai kolom user_name sesuai dengan 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',            -- Data format is JSON
      'scan.startup.mode' = 'earliest-offset' -- Consume 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 from 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 (from 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-value: Pada contoh kode, nilai kolom user_id pada tabel dimensi redis_dim merupakan kunci. Nilai kolom user_name dipetakan ke field user_name, nilai kolom email dipetakan ke field email, dan nilai kolom register_time dipetakan ke field 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',            -- 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 kode, nilai parameter hashName adalah testKey, yang digunakan sebagai kunci. Nilai kolom user_id pada tabel dimensi redis_dim merupakan field, sedangkan nilai kolom user_name merupakan 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',            -- Data format is JSON
      'scan.startup.mode' = 'earliest-offset' -- Consume 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 from 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 (from 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;