Tablestore - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

El conector de Tablestore permite utilizar tablas de Tablestore como tablas de origen, tablas de dimensión y tablas de destino en trabajos de Flink SQL que se ejecutan en modo streaming.

Capacidades del conector

Elemento	Descripción
Modo de ejecución	Modo streaming
Tipo de API	API SQL
Tipo de tabla	Tabla de origen, tabla de dimensión y tabla de destino
Formato de datos	N/D
Métricas de la tabla de destino	`numBytesOut`, `numBytesOutPerSecond`, `numRecordsOut`, `numRecordsOutPerSecond`, `currentSendTime`
Actualización o eliminación de datos en una tabla de destino	Compatible

Para obtener más información sobre las métricas de destino, consulte Métricas de monitoreo .

Requisitos previos

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

Una instancia de Tablestore y una tabla de Tablestore. Consulte Use Tablestore.

Límites de uso

Se admite el acceso entre cuentas a instancias de Tablestore. Al utilizar un endpoint de VPC, la instancia de Tablestore debe encontrarse en la misma región que Flink. Establezca los valores de accessId y accessKey con el par de AccessKey de la cuenta propietaria de la instancia de Tablestore.

Sintaxis

Los tres tipos de tabla utilizan 'connector'='ots' en la cláusula WITH, junto con opciones específicas según el tipo.

Tabla de destino

CREATE TABLE ots_sink (
  name VARCHAR,
  age BIGINT,
  birthday BIGINT,
  PRIMARY KEY (name, age) NOT ENFORCED
) WITH (
  'connector'='ots',
  'instanceName'='<yourInstanceName>',
  'tableName'='<yourTableName>',
  'accessId'='${ak_id}',
  'accessKey'='${ak_secret}',
  'endPoint'='<yourEndpoint>',
  'valueColumns'='birthday'
);

Una tabla de destino de Tablestore requiere una clave primaria. Cada registro de salida se agrega a la tabla para actualizar los datos existentes.

Tabla de dimensión

CREATE TABLE ots_dim (
  id INT,
  len INT,
  content STRING
) WITH (
  'connector'='ots',
  'endPoint'='<yourEndpoint>',
  'instanceName'='<yourInstanceName>',
  'tableName'='<yourTableName>',
  'accessId'='${ak_id}',
  'accessKey'='${ak_secret}'
);

Tabla de origen

CREATE TABLE tablestore_stream (
  `order` VARCHAR,
  orderid VARCHAR,
  customerid VARCHAR,
  customername VARCHAR
) WITH (
  'connector'='ots',
  'endPoint'='<yourEndpoint>',
  'instanceName'='flink-source',
  'tableName'='flink_source_table',
  'tunnelName'='flinksourcestream',
  'accessId'='${ak_id}',
  'accessKey'='${ak_secret}',
  'ignoreDelete'='false'
);

Metadatos disponibles

La tabla de origen de Tablestore expone dos campos de metadatos mediante la palabra clave METADATA. Utilice estos campos para rastrear el tipo de operación y el momento de cada evento de cambio.

Clave de metadatos	Tipo de dato en Flink	Descripción
`type`	STRING	El tipo de operación de datos (se asigna a `OtsRecordType`).
`timestamp`	BIGINT	La hora de la operación de datos en microsegundos (se asigna a `OtsRecordTimestamp`). Se establece en `0` para lecturas completas de datos.

Para leer campos de metadatos, declárelos con la sintaxis METADATA FROM:

CREATE TABLE tablestore_stream (
  `order` VARCHAR,
  orderid VARCHAR,
  customerid VARCHAR,
  customername VARCHAR,
  record_type STRING METADATA FROM 'type',
  record_timestamp BIGINT METADATA FROM 'timestamp'
) WITH (
  ...
);

Opciones del conector

Opciones generales

Todos los tipos de tabla comparten las siguientes opciones.

Opción	Tipo	Obligatorio	Valor predeterminado	Descripción
`connector`	String	Sí	—	Establezca este valor en `ots`.
`instanceName`	String	Sí	—	Nombre de la instancia de Tablestore.
`endPoint`	String	Sí	—	Endpoint de la instancia de Tablestore. Consulte Endpoints.
`tableName`	String	Sí	—	Nombre de la tabla.
`accessId`	String	Sí	—	AccessKey ID de su cuenta de Alibaba Cloud o de un usuario de Resource Access Management (RAM). Consulte How do I view the AccessKey ID and AccessKey secret?
`accessKey`	String	Sí	—	AccessKey secret de su cuenta de Alibaba Cloud o de un usuario de RAM.
`connectTimeout`	Integer	No	30000	Tiempo de espera de conexión en milisegundos.
`socketTimeout`	Integer	No	30000	Tiempo de espera del socket en milisegundos.
`ioThreadCount`	Integer	No	4	Número de hilos de E/S.
`callbackThreadPoolSize`	Integer	No	4	Tamaño del pool de hilos de callback.

Importante

Utilice variables para almacenar su par de AccessKey en lugar de escribirlo directamente en el código.

Opciones de la tabla de origen

Opción	Tipo	Obligatorio	Valor predeterminado	Descripción
`tunnelName`	String	Sí	—	Nombre del túnel de Tablestore. Cree el túnel en la consola de Tablestore antes de usar esta opción. Tipos de túnel compatibles: Incremental, Full y Differential. Consulte la sección "Create a tunnel" en Quick start.
`ignoreDelete`	Boolean	No	false	Indica si se deben omitir las operaciones de eliminación. `true`: omitir; `false`: procesar operaciones de eliminación.
`skipInvalidData`	Boolean	No	false	Indica si se deben omitir los datos no válidos. `true`: omitir datos no válidos; `false`: notificar un error. Requiere Ververica Runtime (VVR) 8.0.4 o posterior.
`retryStrategy`	Enum	No	TIME	Directiva de reintentos. `TIME`: reintentar hasta que transcurra `retryTimeoutMs`; `COUNT`: reintentar hasta alcanzar `retryCount`.
`retryCount`	Integer	No	3	Número máximo de reintentos. Aplica cuando `retryStrategy` es `COUNT`.
`retryTimeoutMs`	Integer	No	180000	Tiempo de espera de reintento en milisegundos. Aplica cuando `retryStrategy` es `TIME`.
`streamOriginColumnMapping`	String	No	—	Asignación de nombres de columna originales a nombres de columna reales. Formato: `origin_col1:col1,origin_col2:col2`.
`outputSpecificRowType`	Boolean	No	false	Indica si se debe transferir el tipo de fila específico. `false`: todas las filas se tratan como INSERT; `true`: las filas pueden ser INSERT, DELETE o UPDATE_AFTER.
`dataFetchTimeoutMs`	Integer	No	10000	Tiempo máximo en milisegundos para obtener datos de una sola partición. Reduzca este valor para disminuir la latencia general de sincronización al sincronizar muchas particiones. Requiere VVR 8.0.10 o posterior.
`enableRequestCompression`	Boolean	No	false	Indica si se debe habilitar la compresión de solicitudes. Reduce el uso de ancho de banda a costa de una mayor carga de CPU. Requiere VVR 8.0.10 o posterior.

Opciones de la tabla de destino

Opción	Tipo	Obligatorio	Valor predeterminado	Descripción
`valueColumns`	String	Sí	—	Nombres de las columnas que se van a escribir. Separe varios nombres de columna con comas (,).
`retryIntervalMs`	Integer	No	1000	Intervalo de reintento en milisegundos.
`maxRetryTimes`	Integer	No	10	Número máximo de reintentos.
`bufferSize`	Integer	No	5000	Número máximo de registros almacenados en búfer antes de desencadenar una escritura.
`batchWriteTimeoutMs`	Integer	No	5000	Tiempo de espera de escritura en milisegundos. Si los registros en búfer no alcanzan `bufferSize` dentro de este periodo, se escriben todos los registros almacenados.
`batchSize`	Integer	No	100	Número de registros escritos por lote. Máximo: 200.
`ignoreDelete`	Boolean	No	false	Indica si se deben omitir las operaciones de eliminación.
`autoIncrementKey`	String	No	—	Nombre de la columna de clave primaria de incremento automático. Configure esta opción solo si la tabla de destino tiene una columna de clave primaria de incremento automático. Requiere VVR 8.0.4 o posterior.
`overwriteMode`	Enum	No	PUT	Modo de escritura. `PUT`: sobrescribir en modo PUT; `UPDATE`: sobrescribir en modo UPDATE. El modo de columna dinámica requiere `UPDATE`.
`defaultTimestampInMillisecond`	Long	No	-1	Marca de tiempo predeterminada para escrituras. Si no se establece, se utiliza la hora actual del sistema.
`dynamicColumnSink`	Boolean	No	false	Indica si se debe habilitar el modo de columna dinámica. En este modo no hay columnas predefinidas; las columnas se insertan según los valores en tiempo de ejecución. Las primeras N columnas definen la clave primaria. La penúltima columna contiene el nombre de la columna y la última contiene su valor; ambas deben ser de tipo STRING. Si se habilita, `overwriteMode` debe ser `UPDATE` y no se admiten claves primarias de incremento automático.
`checkSinkTableMeta`	Boolean	No	true	Indica si se debe comprobar que la clave primaria de la tabla de Tablestore coincide con la clave primaria declarada en la sentencia CREATE TABLE.
`enableRequestCompression`	Boolean	No	false	Indica si se debe habilitar la compresión de solicitudes durante las escrituras.
`maxColumnsCount`	Integer	No	128	Número máximo de columnas escritas en la tabla de destino. Si se establece un valor superior a 128, se produce el error `The count of attribute columns exceeds the maximum`. Requiere VVR 8.0.10 o posterior.
`storageType`	String	No	`WIDE_COLUMN`	Tipo de tabla de destino. `WIDE_COLUMN`: tabla de columnas anchas; `TIMESERIES`: tabla de series temporales.

Opciones de la tabla de dimensión

Funcionamiento de la caché

La caché de la tabla de dimensión reduce las consultas repetidas a Tablestore. Elija una directiva de caché según el tamaño de su tabla y los patrones de consulta:

None: Sin caché. Cada consulta accede directamente a Tablestore. Utilice esta opción cuando los datos cambien con frecuencia y la actualización sea crítica.
LRU: Almacena en caché un número fijo de registros accedidos recientemente. Cuando una consulta falla en la caché, el conector consulta Tablestore y actualiza la caché con el resultado. Establezca cacheSize y cacheTTLMs al usar esta directiva.
ALL (predeterminado): Carga toda la tabla de dimensión en la caché antes de que comience el trabajo. Todas las consultas posteriores se resuelven desde la caché. Cuando la caché caduca (cacheTTLMs), el conector vuelve a cargar todos los datos. Utilice ALL cuando la tabla sea pequeña y espere muchas consultas con claves inexistentes. Al usar ALL, aumente la memoria del nodo de unión, ya que la caché requiere aproximadamente el doble del tamaño de la tabla remota.

Opción	Tipo	Obligatorio	Valor predeterminado	Descripción
`retryIntervalMs`	Integer	No	1000	Intervalo de reintento en milisegundos.
`maxRetryTimes`	Integer	No	10	Número máximo de reintentos.
`cache`	String	No	ALL	Directiva de caché: `None`, `LRU` o `ALL`.
`cacheSize`	Integer	No	—	Número máximo de registros en caché. Aplica cuando `cache` es `LRU`.
`cacheTTLMs`	Integer	No	—	TTL de la caché en milisegundos. Para LRU: tiempo de espera por entrada. Para ALL: intervalo de actualización completa de la caché. Déjelo sin configurar para deshabilitar la caducidad.
`cacheEmpty`	Boolean	No	—	Indica si se deben almacenar en caché los resultados vacíos (sin coincidencias). `true`: almacenar en caché; `false`: no almacenar en caché.
`cacheReloadTimeBlackList`	String	No	—	Ventanas de tiempo durante las cuales no se actualiza la caché ALL. Formato: `2017-10-24 14:00 -> 2017-10-24 15:00, 2017-11-10 23:30 -> 2017-11-11 08:00`. Separe varias ventanas con comas; utilice `->` entre las horas de inicio y fin.
`async`	Boolean	No	false	Indica si se debe habilitar la búsqueda asíncrona. `true`: búsquedas asíncronas (los resultados no están ordenados); `false`: búsquedas síncronas.

Asignaciones de tipos de datos

Tabla de origen

Tipo de Tablestore	Tipo de Flink SQL
INTEGER	BIGINT
STRING	STRING
BOOLEAN	BOOLEAN
DOUBLE	DOUBLE
BINARY	BINARY

Tabla de destino

Tipo de Flink SQL	Tipo de Tablestore
BINARY	BINARY
VARBINARY	BINARY
CHAR	STRING
VARCHAR	STRING
TINYINT	INTEGER
SMALLINT	INTEGER
INTEGER	INTEGER
BIGINT	INTEGER
FLOAT	DOUBLE
DOUBLE	DOUBLE
BOOLEAN	BOOLEAN

Ejemplos

Leer desde Tablestore y escribir en Tablestore

Este ejemplo lee datos de pedidos desde una tabla de origen de Tablestore a través de Tunnel Service y los escribe en una tabla de destino de Tablestore. La tabla de destino utiliza una columna de clave primaria de incremento automático.

CREATE TEMPORARY TABLE tablestore_stream (
  `order` VARCHAR,
  orderid VARCHAR,
  customerid VARCHAR,
  customername VARCHAR
) WITH (
  'connector'='ots',
  'endPoint'='<yourEndpoint>',
  'instanceName'='flink-source',
  'tableName'='flink_source_table',
  'tunnelName'='flinksourcestream',
  'accessId'='${ak_id}',
  'accessKey'='${ak_secret}',
  'ignoreDelete'='false',
  'skipInvalidData'='false'
);

CREATE TEMPORARY TABLE ots_sink (
  `order` VARCHAR,
  orderid VARCHAR,
  customerid VARCHAR,
  customername VARCHAR,
  PRIMARY KEY (`order`, orderid) NOT ENFORCED
) WITH (
  'connector'='ots',
  'endPoint'='<yourEndpoint>',
  'instanceName'='flink-sink',
  'tableName'='flink_sink_table',
  'accessId'='${ak_id}',
  'accessKey'='${ak_secret}',
  'valueColumns'='customerid,customername',
  'autoIncrementKey'='${auto_increment_primary_key_name}'
);

INSERT INTO ots_sink
SELECT `order`, orderid, customerid, customername FROM tablestore_stream;

Sincronizar una tabla de columnas anchas con una tabla de series temporales

Este ejemplo lee desde una tabla de origen de columnas anchas y escribe en una tabla de destino de series temporales. La columna tags de la tabla de destino utiliza MAP<STRING, STRING> para almacenar pares clave-valor de etiquetas, y storageType se establece en TIMESERIES.

CREATE TEMPORARY TABLE timeseries_source (
  measurement STRING,
  datasource STRING,
  tag_a STRING,
  `time` BIGINT,
  binary_value BINARY,
  bool_value BOOLEAN,
  double_value DOUBLE,
  long_value BIGINT,
  string_value STRING,
  tag_b STRING,
  tag_c STRING,
  tag_d STRING,
  tag_e STRING,
  tag_f STRING
) WITH (
  'connector'='ots',
  'endPoint'='https://iotstore-test.cn-hangzhou.vpc.tablestore.aliyuncs.com',
  'instanceName'='iotstore-test',
  'tableName'='test_ots_timeseries_2',
  'tunnelName'='timeseries_source_tunnel_2',
  'accessId'='${ak_id}',
  'accessKey'='${ak_secret}',
  'ignoreDelete'='true'
);

CREATE TEMPORARY TABLE timeseries_sink (
  measurement STRING,
  datasource STRING,
  tags MAP<STRING, STRING>,
  `time` BIGINT,
  binary_value BINARY,
  bool_value BOOLEAN,
  double_value DOUBLE,
  long_value BIGINT,
  string_value STRING,
  tag_b STRING,
  tag_c STRING,
  tag_d STRING,
  tag_e STRING,
  tag_f STRING,
  PRIMARY KEY (measurement, datasource, tags, `time`) NOT ENFORCED
) WITH (
  'connector'='ots',
  'endPoint'='https://iotstore-test.cn-hangzhou.vpc.tablestore.aliyuncs.com',
  'instanceName'='iotstore-test',
  'tableName'='test_timeseries_sink_table_2',
  'accessId'='${ak_id}',
  'accessKey'='${ak_secret}',
  'storageType'='TIMESERIES'
);

-- Build the tags map from individual tag columns and insert into the time series sink table
INSERT INTO timeseries_sink
SELECT
  measurement,
  datasource,
  MAP['tag_a', tag_a, 'tag_b', tag_b, 'tag_c', tag_c, 'tag_d', tag_d, 'tag_e', tag_e, 'tag_f', tag_f] AS tags,
  `time`,
  binary_value,
  bool_value,
  double_value,
  long_value,
  string_value,
  tag_b,
  tag_c,
  tag_d,
  tag_e,
  tag_f
FROM timeseries_source;

Realtime Compute for Apache Flink:Tablestore