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

Background information

Alibaba Cloud Tair is a database service compatible with open source Redis protocols and provides hybrid memory-and-disk storage. Tair uses a high availability (HA) active-active architecture and a scalable cluster architecture to meet business requirements for high throughput, low latency, and flexible scaling up or down. For more information, see What is Tair (Redis OSS-compatible)?.

Prerequisites

• You have created a Tair (Redis OSS-compatible) instance. For more information, see Step 1: Create an instance.

• You have configured a whitelist. For more information, see Step 2: Configure a whitelist.

Limits

• The Redis connector provides best-effort semantics and does not guarantee exactly-once delivery. You must ensure idempotence in your application.

• The following limits apply to dimension tables:

  • Only STRING and HASHMAP data types can be read from Redis.

  • All fields in the dimension table must be of the STRING type. You must declare exactly one primary key.

  • When you join a dimension table, the ON clause must include an equivalent condition for the primary key.

Known issues and solutions

The cache feature in Ververica Runtime (VVR) 8.0.9 has an issue. To disable it, add sink.buffer-flush.max-rows = '0' to the WITH clause of the sink table.

Syntax

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

WITH parameters

Data type formats for Redis sink tables

Type mappings

Usage examples

• Sink table

  • Write STRING data: In the code example, the value from the user_id column in the redis_sink sink table is written to Redis as the key, and the value from the login_time column is written as the value.

    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;

  • Write HASHMAP data in multi-value mode: In the code example, the value from the order_id column in the redis_sink sink table is written to Redis as the key. The values from the product_name, quantity, and amount columns are written to the product_name, quantity, and amount fields, respectively.

    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;

  • Write HASHMAP data in single-value mode: In the code example, the value from the order_id column in the redis_sink sink table is written to Redis as the key. The value from the product_name column is written as the field, and the value from the quantity column is written as the value.

    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;

• Dimension table

  • Read STRING data: In the code example, the value from the user_id column in the redis_dim dimension table corresponds to the key, and the value from the user_name column corresponds to the value.

    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;

  • Read HASHMAP data in multi-value mode: In the code example, the value from the user_id column in the redis_dim dimension table is the key. The value from the user_name column maps to the user_name field. The value from the email column maps to the email field. The value from the register_time column maps to the register_time field.

    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;

  • Read HASHMAP data in single-value mode: In the code example, the value of the hashName parameter is testKey, which is used as the key. The value from the user_id column in the redis_dim dimension table is the field, and the value from the user_name column is the value.

    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;