Tair (Redis OSS-compatible) - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

El conector de Tair (Enterprise Edition) escribe datos de streaming de Flink en una instancia de Tair (Enterprise Edition). Tair es una base de datos compatible con Redis que admite almacenamiento híbrido en memoria y disco, espera activa para alta disponibilidad y una arquitectura de clúster escalable para cargas de trabajo de alto rendimiento y baja latencia.

Funcionalidades compatibles

Categoría	Detalles
Tipos compatibles	Tabla sink
Modo de ejecución	Modo stream
Formato de datos	STRING
Tipo de API	SQL
Actualización o eliminación de datos en una tabla sink	Sí
Métricas únicas	`numBytesSend`, `numBytesSendPerSecond`, `numRecordsSend`, `numRecordsSendPerSecond`, `numRecordSendErrors`, `currentSendTime`

Para obtener más información sobre las métricas, consulte Metrics.

Requisitos previos

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

Una instancia de Tair (Enterprise Edition). Consulte Step 1: Create an instance
Una lista blanca de direcciones IP configurada para la instancia. Consulte Paso 2: Configurar listas blancas

Limitaciones

Realtime Compute for Apache Flink requiere Ververica Runtime (VVR) 6.0.6 o posterior para usar el conector de Tair (Enterprise Edition).
TairTs, TairCpc, TairRoaring, TairVector y TairGis requieren VVR 8.0.1 o posterior.
El conector no admite la configuración de varios hosts.

Sintaxis

La siguiente instrucción DDL crea una tabla sink de Tair. PRIMARY KEY es obligatorio.

CREATE TABLE tair_table (
  a STRING,
  b STRING,
  PRIMARY KEY (a) NOT ENFORCED
) WITH (
  'connector' = 'tair',
  'host'      = '<yourHost>',
  'mode'      = '<dataStructure>'
);

Tair admite todas las estructuras de datos de Redis (STRING, LIST, SET, HASHMAP, SORTEDSET) y sus propias estructuras propietarias. Para ver ejemplos de sintaxis de estructuras de datos compatibles con Redis, consulte Tair (Redis OSS-compatible) connector.

Opciones del conector

Parámetros obligatorios

Parámetro	Tipo	Descripción
`connector`	String	Establezca este valor en `tair`.
`host`	String	Endpoint del servidor de Tair. Utilice un endpoint interno para evitar las limitaciones de latencia y ancho de banda de las conexiones de red pública.
`mode`	STRING	Estructura de datos de destino. Consulte Estructuras de datos compatibles para conocer los valores válidos.

Parámetros de conexión

Parámetro	Tipo	Valor predeterminado	Descripción
`port`	INT	`6379`	Número de puerto del servidor de Tair.
`password`	String	Cadena vacía	Contraseña de la base de datos de Tair. Déjelo en blanco para deshabilitar la verificación por contraseña.
`dbNum`	INT	`0`	ID de la base de datos de destino.
`clusterMode`	BOOLEAN	`false`	Establezca este valor en `true` para usar la arquitectura de clúster. Establézcalo en `false` para el modo independiente.

Parámetros de comportamiento de escritura

Parámetro	Tipo	Valor predeterminado	Descripción
`ignoreDelete`	BOOLEAN	`false`	Controla el comportamiento al recibir un mensaje de retracción. `false`: elimina los datos insertados y su clave. `true`: conserva los datos insertados y su clave.
`incrMode`	STRING	`None`	Modo de escritura del sink. `None`: operación de inserción. `int`: INCRBY con un incremento fijo definido por `incrValue`. `float`: INCRBYFLOAT con un incremento fijo definido por `incrValue`. `dynamic_int`: INCRBY con el incremento leído de la columna indicada por `incrValue`. `dynamic_float`: INCRBYFLOAT con el incremento leído de la columna indicada por `incrValue`.
`incrValue`	STRING	Ninguna	Cuando `incrMode` es `int` o `float`, representa el valor de incremento fijo. Cuando `incrMode` es `dynamic_int` o `dynamic_float`, corresponde al nombre de la columna DDL que contiene el valor de incremento. No se utiliza cuando `incrMode` es `None`.

Parámetros de caducidad de claves

Parámetro	Tipo	Valor predeterminado	Descripción
`expiration`	LONG	`0`	Tiempo de vida (TTL) relativo para las claves insertadas, en milisegundos. `0` deshabilita el TTL.
`expirationAt`	LONG	`0`	Marca de tiempo de caducidad absoluta para las claves insertadas, en milisegundos. Solo surte efecto cuando `expiration` se establece en `0`. `0` deshabilita la caducidad absoluta.

Parámetros de caducidad de campos (solo TairHash y TairTs)

Parámetro	Tipo	Valor predeterminado	Descripción
`fieldExpireMode`	String	`None`	Modo de caducidad para campos en TairHash o skeys en TairTs. `None`: sin caducidad. `millisecond`: caducidad relativa establecida según el valor de `fieldExpireValue`. `unixtime`: caducidad absoluta establecida según el valor de `fieldExpireValue`. `dynamic_millisecond`: caducidad relativa leída de la columna indicada por `fieldExpireValue`. `dynamic_unixtime`: caducidad absoluta leída de la columna indicada por `fieldExpireValue`. Importante Las skeys de TairTs deben usar `millisecond`.
`fieldExpireValue`	String	Ninguna	Cuando `fieldExpireMode` es `millisecond` o `unixtime`, representa el valor de caducidad fijo. Cuando `fieldExpireMode` es `dynamic_millisecond` o `dynamic_unixtime`, corresponde al nombre de la columna DDL que contiene el valor de caducidad.

Mapeo de tipos de datos

Tipo de Flink	Tipo de Tair
VARCHAR	STRING
DOUBLE	DOUBLE

Estructuras de datos compatibles

El parámetro mode acepta los siguientes valores. Cada estructura de datos tiene requisitos específicos de columnas DDL que varían según incrMode.

Estructuras de datos compatibles con Redis

Para consultar los formatos DDL de STRING, LIST, SET, HASHMAP y SORTEDSET, véase Tair (Redis OSS-compatible) connector.

Estructuras de datos propietarias de Tair

Estructura de datos	`incrMode`	Columnas DDL	Comando de escritura
TairString	`None`	2: key (STRING), value (STRING)	`exset key value`
TairString	`int` o `float`	1: key (STRING)	`exincrby/exincrbyfloat key incrValue`
TairString	`dynamic_int` o `dynamic_float`	2: key (STRING), incrValue (STRING)	`exincrby/exincrbyfloat key incrValue`
TairHash	`None`	3: key (STRING), field (STRING), value (STRING)	`exhset key field value`
TairHash	`int` o `float`	2: key (STRING), field (STRING)	`exhincrby/exincrbyfloat key field incrValue`
TairHash	`dynamic_int` o `dynamic_float`	3: key (STRING), field (STRING), incrValue (STRING)	`exhincrby/exincrbyfloat key field incrValue`
TairZset	`None`	3–258: key (STRING), member (STRING), score×N (DOUBLE, hasta 256 dimensiones)	`exzadd key score member`
TairZset	`int` o `float`	2: key (STRING), member (STRING)	`exzincyby key member incrValue`
TairZset	`dynamic_int` o `dynamic_float`	3: key (STRING), member (STRING), incrValue (STRING)	`exzincyby key member incrValue`
TairBloom	Debe ser `None`	2: key (STRING), item (STRING)	`BF.ADD key item`
TairDoc	Debe ser `None`	3: key (STRING), path (STRING), json (STRING)	`JSON.SET key path json`
TairSearch	`None`	4: index (STRING), doc_id (STRING), document (STRING, JSON), mapping (STRING)	`TFT.ADDDOC index document docid`
TairSearch	`int` o `float`	4: index (STRING), doc_id (STRING), field (STRING), mapping (STRING)	`TFT.INCRLONGDOCFIELD/TFT.INCRFLOATDOCFIELD index doc_id field increment`
TairSearch	`dynamic_int` o `dynamic_float`	5: index (STRING), doc_id (STRING), field (STRING), mapping (STRING), incrValue (STRING)	`TFT.INCRLONGDOCFIELD/TFT.INCRFLOATDOCFIELD index doc_id field increment`
TairCpc	Debe ser `None`	2: key (STRING), item (STRING)	`CPC.UPDATE key item`
TairGis	Debe ser `None`	3: key (STRING), polygon_name (STRING), polygon_wkt (STRING)	`GIS.ADD area polygonName polygonWkt`
TairRoaring	Debe ser `None`	3: key (STRING), offset (BIGINT), value (BIGINT, 0 o 1)	`TR.SETBIT key offset value`
TairVector	Debe ser `None`	6: index_name (STRING), pk (STRING), vector_data (STRING), dims (INT), algorithm (STRING), distance_method (STRING)	`TVS.HSET index_name key VECTOR vector_data`
TairTs	`None`	4: pkey (STRING, grupo de línea temporal), skey (STRING, línea temporal individual), timestamp (STRING), value (STRING)	`EXTS.S.RAW_MODIFY Pkey Skey timestamp value`
TairTs	`float`	3: pkey (STRING), skey (STRING), timestamp (STRING)	`EXTS.S.RAW_INCRBY Pkey Skey timestamp incrValue`
TairTs	`dynamic_float`	4: pkey (STRING), skey (STRING), timestamp (STRING), incrValue (STRING)	`EXTS.S.RAW_INCRBY Pkey Skey timestamp incrValue`

TairSearch y TairVector requieren que se cree un índice y un mapeo antes de insertar datos. Para TairSearch, ejecute TFT.CREATEINDEX index mappings . Para TairVector, ejecute TVS.CREATEINDEX index_name dims algorithm distance_method . TairBloom crea una clave con una capacidad predeterminada de 100 elementos y una tasa de error de 0.01 en la primera inserción. Para la ordenación multidimensional de TairZset, todas las dimensiones de puntuación deben usar el mismo formato.

Ejemplos

Escritura de datos en una tabla sink de TairSearch

Este ejemplo utiliza index_name como índice de TairSearch, doc_id como ID del documento, doc como cuerpo del documento JSON y mapping como definición del mapeo del índice.

CREATE TEMPORARY TABLE datagen_stream (
  v STRING,   -- document content
  p STRING    -- mapping definition
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE tair_output (
  index_name STRING,  -- TairSearch index name
  doc_id     STRING,  -- document ID
  doc        STRING,  -- document body (JSON)
  mapping    STRING,  -- index mapping
  PRIMARY KEY (index_name) NOT ENFORCED
) WITH (
  'connector' = 'tair',
  'mode'      = 'tairsearch',
  'host'      = '${tairHost}',
  'port'      = '${tairPort}',
  'password'  = '${password}'
);

INSERT INTO tair_output
SELECT
  'index' AS index_name,
  v AS doc_id,
  p AS doc,
  '{"mappings":{"_source":{"enabled":true},"properties":{"product_id":{"type":"keyword","ignore_above":128},"product_name":{"type":"text"}}}}' AS mapping
FROM datagen_stream;

Escritura de datos con incremento dinámico (TairString con `incrMode=dynamic_float`)

Este ejemplo usa key como clave de TairString y step como valor de incremento por registro, leído de la columna step.

CREATE TEMPORARY TABLE datagen_stream (
  v STRING,   -- key
  p STRING    -- increment value per record
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE tair_output (
  key  STRING,  -- TairString key
  step STRING,  -- increment value (column referenced by incrValue)
  PRIMARY KEY (key) NOT ENFORCED
) WITH (
  'connector' = 'tair',
  'mode'      = 'tairstring',
  'host'      = '${tairHost}',
  'port'      = '${tairPort}',
  'password'  = '${password}',
  'incrMode'  = 'dynamic_float',
  'incrValue' = 'step'
);

INSERT INTO tair_output
SELECT *
FROM datagen_stream;

Escritura de datos con incremento fijo (TairString con `incrMode=float`)

Este ejemplo aplica un incremento fijo de 11.11 a cada clave en cada operación de escritura.

CREATE TEMPORARY TABLE datagen_stream (
  v STRING,
  p STRING
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE tair_output (
  key STRING,  -- TairString key
  PRIMARY KEY (key) NOT ENFORCED
) WITH (
  'connector' = 'tair',
  'mode'      = 'tairstring',
  'host'      = '${tairHost}',
  'port'      = '${tairPort}',
  'password'  = '${password}',
  'incrMode'  = 'float',
  'incrValue' = '11.11'
);

INSERT INTO tair_output
SELECT v
FROM datagen_stream;

Escritura de datos con TTL a nivel de campo (TairHash con `fieldExpireMode=millisecond`)

Este ejemplo escribe en una tabla sink de TairHash donde cada campo caduca 1000 milisegundos después de su escritura.

CREATE TEMPORARY TABLE datagen_stream (
  v STRING,
  p STRING,
  s STRING
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE tair_output (
  key   STRING,  -- TairHash key
  field STRING,  -- hash field
  value STRING,  -- field value
  PRIMARY KEY (key) NOT ENFORCED
) WITH (
  'connector'        = 'tair',
  'mode'             = 'tairhash',
  'host'             = '${tairHost}',
  'port'             = '${tairPort}',
  'password'         = '${password}',
  'fieldExpireMode'  = 'millisecond',
  'fieldExpireValue' = '1000'
);

INSERT INTO tair_output
SELECT v, p, s
FROM datagen_stream;

Pasos siguientes

Tair (Redis OSS-compatible) connector — ejemplos de sintaxis para estructuras de datos compatibles con Redis
Metrics — lista completa de métricas del conector y sus definiciones