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

El conector de Tair (compatible con Redis OSS) permite leer datos de Tair como tabla de dimensiones y escribir en Tair como tabla destino en trabajos de streaming de Flink SQL.

Descripción general del conector

Elemento	Detalles
Tipos de tabla	Tabla de dimensiones, tabla destino
Modo de ejecución	Streaming
Formato de datos	STRING
Métricas	Tablas de dimensiones: ninguna. Tablas destino: `numBytesOut`, `numRecordsOutPerSecond`, `numBytesOutPerSecond`, `currentSendTime`. Para más información, consulte Métricas de monitoreo.
Tipo de API	API de SQL
Actualizaciones y eliminaciones de datos en tablas destino	Compatible

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Una instancia de Tair (compatible con Redis OSS). Consulte Step 1: Create an instance.
Una lista de permitidos configurada para la instancia. Consulte Step 2: Configure whitelists.

Limitaciones

Semántica de entrega: El conector solo admite entrega best-effort. No se admite la semántica exactly-once. Asegúrese de que sus operaciones de escritura sean idempotentes.
Tipos de datos de tablas de dimensiones: Las tablas de dimensiones solo pueden leer datos STRING y HASHMAP. Todos los campos deben ser de tipo STRING.
Clave primaria de tabla de dimensiones: Cada tabla de dimensiones debe tener exactamente una clave primaria. La cláusula ON en un JOIN de tabla de dimensiones debe usar condiciones de igualdad sobre la clave primaria.

Problemas conocidos

VVR 8.0.9 — Bug de caché en Buffered Writer: Existe un problema con la caché de Buffered Writer en VVR 8.0.9. Como solución temporal, establezca sink.buffer-flush.max-rows en 0 dentro de la cláusula WITH de la tabla destino.

Sintaxis

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.
);

Opciones del conector

Opciones generales

Estas opciones aplican tanto a tablas destino como a tablas de dimensiones.

Opción	Tipo de dato	Obligatorio	Predeterminado	Descripción
`connector`	STRING	Sí	—	Establezca este valor en `redis`.
`host`	STRING	Sí	—	Dirección IP utilizada para conectarse a la base de datos de ApsaraDB for Redis. Utilice el endpoint interno siempre que sea posible. Las conexiones por Internet pueden experimentar mayor latencia o limitaciones de ancho de banda.
`port`	INT	No	`6379`	Número de puerto.
`password`	STRING	No	(cadena vacía)	Contraseña de acceso.
`dbNum`	INT	No	`0`	Número de secuencia de la base de datos.
`clusterMode`	BOOLEAN	No	`false`	Indica si la base de datos está en modo clúster.
`hostAndPorts`	STRING	No	—	Pares de host y puerto con el formato `"host1:port1,host2:port2"`. Es obligatorio cuando `clusterMode` es `true` y se requiere alta disponibilidad (HA) para las conexiones Jedis al clúster de Redis autogestionado. Tiene prioridad sobre `host` y `port`. Si `clusterMode` es `true` pero no se necesita HA, puede configurar únicamente `host` y `port` para especificar un solo nodo.
`key-prefix`	STRING	No	—	Prefijo que se agrega al valor de la clave primaria al leer desde una tabla de dimensiones o escribir en una tabla destino. Un delimitador especificado mediante `key-prefix-delimiter` separa el prefijo del valor de la clave primaria. Requiere VVR 8.0.7 o posterior.
`key-prefix-delimiter`	STRING	No	—	Delimitador entre el prefijo de la clave y el valor de la clave primaria.
`connection.pool.max-total`	INT	No	`8`	Número máximo de conexiones que el pool de conexiones puede asignar. Requiere VVR 8.0.9 o posterior.
`connection.pool.max-idle`	INT	No	`8`	Número máximo de conexiones inactivas en el pool de conexiones.
`connection.pool.min-idle`	INT	No	`0`	Número mínimo de conexiones inactivas en el pool de conexiones.
`connection.pool.lifo`	Boolean	No	`true`	Indica si las conexiones inactivas se asignan desde el pool de conexiones en orden LIFO. Valores válidos: true: LIFO. El pool asigna primero la conexión devuelta más recientemente. false: FIFO. Se asigna primero la conexión inactiva que lleva más tiempo sin usarse. En una instancia de clúster de Redis detrás de un proxy, establecer esta opción en false permite distribuir la carga entre los proxies de manera más uniforme y evitar desequilibrios en las conexiones. Sin embargo, el modo FIFO tiende a mantener las conexiones ocupadas y puede incrementar el número total de conexiones; evalúe esto con cuidado en escenarios con muchas conexiones. Nota: Esta opción solo es compatible con el motor Realtime Compute VVR 11.8.0 y versiones posteriores.
`connect.timeout`	DURATION	No	`3000ms`	Tiempo de espera para establecer la conexión.
`socket.timeout`	DURATION	No	`3000ms`	Tiempo de espera para recibir datos del servidor de Redis.
`cacert.filepath`	STRING	No	—	Ruta completa al certificado SSL/TLS. El archivo debe estar en formato JKS. Si no se configura, el cifrado SSL/TLS permanece deshabilitado. Para habilitar el cifrado, download the CA certificate y upload it as an additional dependency; el archivo se almacena en el directorio `/flink/usrlib`. Ejemplo: `'cacert.filepath' = '/flink/usrlib/ca.jks'`. Requiere VVR 11.1 o posterior.

Opciones de tabla destino

Opción	Tipo de dato	Obligatorio	Predeterminado	Descripción
`mode`	STRING	Sí	—	Estructura de datos de Redis para la tabla destino. Se admiten cinco estructuras: STRING, LIST, SET, HASHMAP y SORTEDSET. La sentencia DDL debe coincidir con la estructura elegida. Consulte Data structures for sink tables.
`flattenHash`	BOOLEAN	No	`false`	Indica si se escriben datos HASHMAP en modo multivalor. Cuando es `true`, declare varios campos que no sean clave primaria: la clave primaria se asigna a la clave de Redis, cada nombre de campo no primario se asigna a un campo Hash y cada valor de campo se asigna al valor Hash correspondiente. Cuando es `false` (modo de valor único), declare exactamente tres campos: la clave primaria se asigna a la clave, el primer campo no primario se asigna al campo Hash y el segundo al valor Hash. Solo surte efecto cuando `mode` es `HASHMAP`. Requiere VVR 8.0.7 o posterior.
`ignoreDelete`	BOOLEAN	No	`false`	Indica si se ignoran los mensajes de retractación. Cuando es `true`, los mensajes de retractación se descartan. Cuando es `false`, la clave y sus datos se eliminan al recibir un mensaje de retractación.
`expiration`	LONG	No	`0`	Tiempo de vida (TTL) de las claves insertadas, en milisegundos. El valor `0` deshabilita el TTL.
`sink.buffer-flush.max-rows`	INT	No	`200`	Número máximo de registros (eventos de inserción, modificación y eliminación) que se retienen en el búfer antes de vaciarlo. Requiere VVR 8.0.9 o posterior cuando `clusterMode = false`; requiere VVR 11.4.0 o posterior cuando `clusterMode = true`.
`sink.buffer-flush.interval`	DURATION	No	`1000ms`	Intervalo en el que el búfer se vacía de forma asíncrona. Requiere VVR 8.0.9 o posterior cuando `clusterMode = false`; requiere VVR 11.4.0 o posterior cuando `clusterMode = true`.

Opciones de tabla de dimensiones

Opción	Tipo de dato	Obligatorio	Predeterminado	Descripción
`mode`	STRING	No	`STRING`	Tipo de dato que se lee desde la tabla de dimensiones. `STRING` lee datos STRING. `HASHMAP` lee datos Hash anidados (Clave → Map\<Campo, Valor\>): declare varias columnas que no sean clave primaria, donde la clave primaria se asigna a la clave de Redis, cada nombre de columna no primaria se asigna a un campo Hash y su valor se asigna al valor del campo. Requiere VVR 8.0.7 o posterior. Para leer datos HASHMAP en modo de valor único, configure `hashName` en su lugar.
`hashName`	STRING	No	—	Clave Hash fija que se utiliza al leer datos HASHMAP en modo de valor único. Al configurarla, declare dos campos: la clave primaria se asigna al campo Hash y la clave no primaria se asigna al valor Hash.
`cache`	STRING	No	`None`	Directiva de caché. `None` deshabilita el almacenamiento en caché. `LRU` almacena en caché un subconjunto de datos; si ocurre un fallo de caché, el conector consulta la tabla de dimensiones. `ALL` carga toda la tabla de dimensiones en la caché antes de ejecutar el despliegue; todas las consultas posteriores usan la caché, la cual se recarga cuando expiran las entradas. Consulte Notas de uso para la opción de caché.
`cacheSize`	LONG	No	`10000`	Número máximo de filas que se almacenan en caché. Es obligatorio cuando `cache` es `LRU`.
`cacheTTLMs`	LONG	No	—	Tiempo de espera de la caché en milisegundos. Para `LRU`, define la expiración por entrada (sin expiración de forma predeterminada). Para `ALL`, define el intervalo de recarga de la caché (sin recarga de forma predeterminada). No tiene efecto cuando `cache` es `None`.
`cacheEmpty`	BOOLEAN	No	`true`	Indica si se almacenan en caché los resultados vacíos (sin coincidencias).
`cacheReloadTimeBlackList`	STRING	No	—	Períodos durante los cuales la directiva de caché `ALL` no realiza recargas. Resulta útil durante eventos de alto tráfico. Formato: utilice `->` para separar las horas de inicio y fin, y `,` para separar varios períodos. Ejemplos: un solo día `2017-10-24 14:00 -> 2017-10-24 15:00`; entre días `2017-11-10 23:30 -> 2017-11-11 08:00`; recurrente diario `12:00 -> 14:00, 22:00 -> 2:00` (requiere VVR 11.1 o posterior).
`async`	BOOLEAN	No	`false`	Indica si se habilita la búsqueda asíncrona. Cuando es `true`, los resultados se devuelven sin garantizar el orden.

Notas de uso para la opción de caché

ALL requiere VVR 8.0.3 o posterior.
Para versiones desde VVR 8.0.3 hasta VVR 11.1 (exclusive), cache = ALL lee HASHMAP únicamente en modo de valor único. En la cláusula WITH del DDL, establezca hashName con el nombre de la clave; declare Campo como clave primaria y Valor como columna no primaria.
A partir de VVR 11.1, cache = ALL admite el modo multivalor de HASHMAP. Especifique la clave de Redis como clave primaria y declare varias columnas no primarias para cada campo Hash. Establezca mode = HASHMAP en la cláusula WITH.
La opción cache debe utilizarse junto con cacheSize y cacheTTLMs.

Estructuras de datos para tablas destino

Cada estructura de datos de Redis corresponde a un esquema DDL y un comando de escritura específicos.

Estructura de datos	Esquema DDL	Comando de escritura
STRING	Dos columnas: clave (STRING), valor (STRING)	`set key value`
LIST	Dos columnas: clave (STRING), valor (STRING)	`lpush key value`
SET	Dos columnas: clave (STRING), valor (STRING)	`sadd key value`
HASHMAP (modo de valor único, predeterminado)	Tres columnas: clave (STRING), campo (STRING), valor (STRING)	`hmset key field value`
HASHMAP (modo multivalor, `flattenHash = true`)	Varias columnas: clave (STRING) y luego una columna por cada campo Hash; cada nombre de columna es el nombre del campo y su valor es el valor del campo	`hmset key col1 value1 col2 value2 ...`
SORTEDSET	Tres columnas: clave (STRING), puntuación (DOUBLE), valor (STRING)	`zadd key score value`

La opción ignoreDelete controla cómo se manejan los mensajes de retractación. Si se establece en true , se omiten las operaciones de eliminación.

Mapeo de tipos de datos

Ámbito	Tipo de Tair (ApsaraDB for Redis)	Tipo de Flink
Todos los tipos de tabla	STRING	STRING
Solo tablas destino	SCORE	DOUBLE

El tipo SCORE se utiliza con datos SORTEDSET. Cada valor en un conjunto ordenado requiere una puntuación DOUBLE, y los valores se ordenan de forma ascendente según dicha puntuación.

Ejemplos

Ejemplos de tablas destino

Todos los ejemplos de tablas destino leen desde un origen Kafka y escriben en un destino Tair.

Escritura de datos STRING

Este ejemplo utiliza user_id como clave de Redis y login_time como valor de Redis.

CREATE TEMPORARY TABLE kafka_source (
  user_id    STRING,   -- User ID
  login_time STRING    -- Login time (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,   -- Redis key
  login_time STRING,   -- Redis value
  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;

Escritura de datos HASHMAP en modo multivalor

Este ejemplo utiliza order_id como clave de Redis y escribe product_name, quantity y amount como campos Hash independientes.

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',
  'properties.bootstrap.servers'  = '<yourKafkaBroker>',
  'format'                        = 'json',
  'scan.startup.mode'             = 'earliest-offset'
);

CREATE TEMPORARY TABLE redis_sink (
  order_id     STRING,  -- Redis key
  product_name STRING,  -- Hash field: product_name
  quantity     STRING,  -- Hash field: quantity
  amount       STRING,  -- Hash field: 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;

Escritura de datos HASHMAP en modo de valor único

Este ejemplo utiliza order_id como clave de Redis, product_name como campo Hash y quantity como valor Hash.

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',
  'properties.bootstrap.servers'  = '<yourKafkaBroker>',
  'format'                        = 'json',
  'scan.startup.mode'             = 'earliest-offset'
);

CREATE TEMPORARY TABLE redis_sink (
  order_id     STRING,  -- Redis key
  product_name STRING,  -- Redis Hash field
  quantity     STRING,  -- Redis Hash 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;

Ejemplos de tablas de dimensiones

Todos los ejemplos de tablas de dimensiones buscan información de usuarios en una tabla de dimensiones Tair y la combinan con un stream de Kafka.

Lectura de datos STRING

Este ejemplo utiliza user_id como clave de Redis y recupera user_name como valor de 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 key
  user_name STRING,   -- Redis value
  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;

Lectura de datos HASHMAP en modo multivalor

Este ejemplo utiliza user_id como clave de Redis y recupera varios campos Hash —user_name, email y register_time— en una sola búsqueda.

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 key
  user_name     STRING,   -- Hash field: user_name
  email         STRING,   -- Hash field: email
  register_time STRING,   -- Hash field: 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;

Lectura de datos HASHMAP en modo de valor único

Este ejemplo utiliza una clave Hash fija (testkey) definida mediante hashName. La columna user_id se asigna al campo Hash y user_name se asigna al valor 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,   -- Hash field
  user_name STRING,   -- Hash value
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'redis',
  'host'      = '<yourHost>',
  'port'      = '<yourPort>',
  'password'  = '<yourPassword>',
  'hashName'  = 'testkey'   -- Fixed Hash key
);

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;