Tablestore - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Pré-requisitos

Antes de começar, certifique-se de ter:

Uma instância do Tablestore e uma tabela do Tablestore. Consulte Usar o Tablestore.

Limites de uso

O acesso entre contas a instâncias do Tablestore é suportado. Ao usar um endpoint de VPC, a instância do Tablestore deve estar na mesma região que o Flink. Defina accessId e accessKey como o par de AccessKey da conta proprietária da instância do Tablestore.

Sintaxe

Os três tipos de tabela usam 'connector'='ots' na cláusula WITH, com opções específicas para cada tipo.

Tabela 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'
);

Uma tabela de destino do Tablestore requer uma chave primária. Cada registro de saída é acrescentado à tabela para atualizar os dados existentes.

Tabela de dimensão

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}'
);

Tabela de origem

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

Metadados disponíveis

A tabela de origem do Tablestore expõe dois campos de metadados por meio da palavra-chave METADATA. Use esses campos para rastrear o tipo de operação e o momento de cada evento de alteração.

Chave de metadado	Tipo de dados do Flink	Descrição
`type`	STRING	O tipo de operação de dados (mapeia para `OtsRecordType`).
`timestamp`	BIGINT	O horário da operação de dados em microssegundos (mapeia para `OtsRecordTimestamp`). Defina como `0` para leituras de dados completos.

Para ler os campos de metadados, declare-os com a sintaxe 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 (
  ...
);

Opções do conector

Opções gerais

Todos os tipos de tabela compartilham as seguintes opções.

Opção	Tipo	Obrigatório	Padrão	Descrição
`connector`	String	Sim	—	Defina como `ots`.
`instanceName`	String	Sim	—	Nome da instância do Tablestore.
`endPoint`	String	Sim	—	Endpoint da instância do Tablestore. Consulte Endpoints.
`tableName`	String	Sim	—	Nome da tabela.
`accessId`	String	Sim	—	AccessKey ID da sua conta Alibaba Cloud ou de um usuário do Resource Access Management (RAM). Consulte Como visualizo o AccessKey ID e o AccessKey secret?
`accessKey`	String	Sim	—	AccessKey secret da sua conta Alibaba Cloud ou de um usuário do RAM.
`connectTimeout`	Integer	Não	30000	Tempo limite de conexão em milissegundos.
`socketTimeout`	Integer	Não	30000	Tempo limite do socket em milissegundos.
`ioThreadCount`	Integer	Não	4	Número de threads de I/O.
`callbackThreadPoolSize`	Integer	Não	4	Tamanho do pool de threads de callback.

Importante

Use variáveis para armazenar seu par de AccessKey em vez de codificá-lo diretamente.

Opções da tabela de origem

Opção	Tipo	Obrigatório	Padrão	Descrição
`tunnelName`	String	Sim	—	Nome do túnel do Tablestore. Crie o túnel no console do Tablestore antes de usar esta opção. Tipos de túnel suportados: Incremental, Completo e Diferencial. Consulte a seção "Criar um túnel" em Guia de início rápido.
`ignoreDelete`	Boolean	Não	false	Se deve ignorar operações de exclusão. `true`: ignorar; `false`: processar operações de exclusão.
`skipInvalidData`	Boolean	Não	false	Se deve ignorar dados inconsistentes. `true`: ignorar dados inconsistentes; `false`: reportar um erro. Requer Ververica Runtime (VVR) 8.0.4 ou posterior.
`retryStrategy`	Enum	Não	TIME	Política de tentativa. `TIME`: tentar até `retryTimeoutMs` expirar; `COUNT`: tentar até `retryCount` ser atingido.
`retryCount`	Integer	Não	3	Número máximo de tentativas. Aplica-se quando `retryStrategy` é `COUNT`.
`retryTimeoutMs`	Integer	Não	180000	Tempo limite de tentativa em milissegundos. Aplica-se quando `retryStrategy` é `TIME`.
`streamOriginColumnMapping`	String	Não	—	Mapeamento de nomes de colunas originais para nomes de colunas reais. Formato: `origin_col1:col1,origin_col2:col2`.
`outputSpecificRowType`	Boolean	Não	false	Se deve propagar o tipo de linha específico. `false`: todas as linhas são tratadas como INSERT; `true`: as linhas podem ser INSERT, DELETE ou UPDATE_AFTER.
`dataFetchTimeoutMs`	Integer	Não	10000	Tempo máximo em milissegundos para buscar dados de uma única partição. Reduza este valor para diminuir a latência geral de sincronização ao sincronizar muitas partições. Requer VVR 8.0.10 ou posterior.
`enableRequestCompression`	Boolean	Não	false	Se deve ativar a compressão de requisições. Reduz o uso de largura de banda ao custo de maior carga de CPU. Requer VVR 8.0.10 ou posterior.

Opções da tabela de destino

Opção	Tipo	Obrigatório	Padrão	Descrição
`valueColumns`	String	Sim	—	Nomes das colunas a serem gravadas. Separe vários nomes de coluna com vírgulas (,).
`retryIntervalMs`	Integer	Não	1000	Intervalo de tentativa em milissegundos.
`maxRetryTimes`	Integer	Não	10	Número máximo de tentativas.
`bufferSize`	Integer	Não	5000	Número máximo de registros no buffer antes que uma gravação seja acionada.
`batchWriteTimeoutMs`	Integer	Não	5000	Tempo limite de gravação em milissegundos. Se os registros no buffer não atingirem `bufferSize` dentro desse período, todos os registros no buffer serão gravados.
`batchSize`	Integer	Não	100	Número de registros gravados por lote. Máximo: 200.
`ignoreDelete`	Boolean	Não	false	Se deve ignorar operações de exclusão.
`autoIncrementKey`	String	Não	—	Nome da coluna de chave primária com incremento automático. Configure apenas se a tabela de destino tiver uma coluna de chave primária com incremento automático. Requer VVR 8.0.4 ou posterior.
`overwriteMode`	Enum	Não	PUT	Modo de gravação. `PUT`: sobrescrever no modo PUT; `UPDATE`: sobrescrever no modo UPDATE. O modo de coluna dinâmica requer `UPDATE`.
`defaultTimestampInMillisecond`	Long	Não	-1	Timestamp padrão para gravações. Se não definido, a hora atual do sistema é usada.
`dynamicColumnSink`	Boolean	Não	false	Se deve ativar o modo de coluna dinâmica. Nesse modo, nenhuma coluna é predefinida; as colunas são inseridas com base nos valores de execução. As primeiras N colunas definem a chave primária. A penúltima coluna contém o nome da coluna e a última coluna contém seu valor — ambas devem ser STRING. Se ativado, `overwriteMode` deve ser `UPDATE` e chaves primárias com incremento automático não são suportadas.
`checkSinkTableMeta`	Boolean	Não	true	Se deve verificar se a chave primária da tabela do Tablestore corresponde à chave primária declarada na instrução CREATE TABLE.
`enableRequestCompression`	Boolean	Não	false	Se deve ativar a compressão de requisições durante gravações.
`maxColumnsCount`	Integer	Não	128	Número máximo de colunas gravadas na tabela de destino. Se definido acima de 128, o erro `The count of attribute columns exceeds the maximum` ocorre. Requer VVR 8.0.10 ou posterior.
`storageType`	String	Não	`WIDE_COLUMN`	Tipo da tabela de destino. `WIDE_COLUMN`: tabela de colunas largas; `TIMESERIES`: tabela de séries temporais.

Opções da tabela de dimensão

Como o cache funciona

O cache da tabela de dimensão reduz consultas repetidas ao Tablestore. Escolha uma política de cache com base no tamanho da sua tabela e nos padrões de consulta:

None: Sem cache. Cada consulta acessa o Tablestore diretamente. Use quando os dados mudam frequentemente e a atualização é crítica.
LRU: Armazena em cache um número fixo de registros acessados recentemente. Quando uma consulta não encontra o dado no cache, o conector consulta o Tablestore e atualiza o cache com o resultado. Defina cacheSize e cacheTTLMs ao usar esta política.
ALL (padrão): Carrega toda a tabela de dimensão no cache antes do início do job. Todas as consultas subsequentes são atendidas pelo cache. Quando o cache expira (cacheTTLMs), o conector recarrega todos os dados. Use ALL quando a tabela é pequena e você espera muitas consultas com chaves ausentes. Ao usar ALL, aumente a memória do nó de junção — o cache requer aproximadamente o dobro do tamanho da tabela remota.

Opção	Tipo	Obrigatório	Padrão	Descrição
`retryIntervalMs`	Integer	Não	1000	Intervalo de tentativa em milissegundos.
`maxRetryTimes`	Integer	Não	10	Número máximo de tentativas.
`cache`	String	Não	ALL	Política de cache: `None`, `LRU` ou `ALL`.
`cacheSize`	Integer	Não	—	Número máximo de registros em cache. Aplica-se quando `cache` é `LRU`.
`cacheTTLMs`	Integer	Não	—	TTL do cache em milissegundos. Para LRU: tempo limite por entrada. Para ALL: intervalo de atualização completa do cache. Deixe sem definir para desativar a expiração.
`cacheEmpty`	Boolean	Não	—	Se deve armazenar resultados vazios (sem correspondência) em cache. `true`: armazenar em cache; `false`: não armazenar em cache.
`cacheReloadTimeBlackList`	String	Não	—	Janelas de tempo durante as quais o cache ALL não é atualizado. Formato: `2017-10-24 14:00 -> 2017-10-24 15:00, 2017-11-10 23:30 -> 2017-11-11 08:00`. Separe várias janelas com vírgulas; use `->` entre os horários de início e fim.
`async`	Boolean	Não	false	Se deve ativar a consulta assíncrona. `true`: consultas assíncronas (os resultados não são ordenados); `false`: consultas síncronas.

Mapeamentos de tipos de dados

Tabela de origem

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

Tabela de destino

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

Exemplos

Ler do Tablestore e gravar no Tablestore

Este exemplo lê dados de pedidos de uma tabela de origem do Tablestore por meio do Tunnel Service e os grava em uma tabela de destino do Tablestore. A tabela de destino usa uma coluna de chave primária com 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 uma tabela de colunas largas para uma tabela de séries temporais

Este exemplo lê de uma tabela de origem de colunas largas e grava em uma tabela de destino de séries temporais. A coluna tags da tabela de destino usa MAP<STRING, STRING> para armazenar pares de chave-valor de tags, e storageType está definido como 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;

Item	Descrição
Modo de execução	Modo streaming
Tipo de API	API SQL
Tipo de tabela	Tabela de origem, tabela de dimensão e tabela de destino
Formato de dados	N/A
Métricas da tabela de destino	`numBytesOut`, `numBytesOutPerSecond`, `numRecordsOut`, `numRecordsOutPerSecond`, `currentSendTime`
Atualização ou exclusão de dados na tabela de destino	Suportado

Realtime Compute for Apache Flink:Tablestore