Conector Tair (compatível com Redis OSS) - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Visão geral do conector

Item	Detalhes
Tipos de tabela	Tabela de dimensão, tabela sink
Modo de execução	Streaming
Formato de dados	STRING
Métricas	Tabelas de dimensão: nenhuma. Tabelas sink: `numBytesOut`, `numRecordsOutPerSecond`, `numBytesOutPerSecond`, `currentSendTime`. Para mais detalhes, consulte Métricas de monitoramento.
Tipo de API	API SQL
Atualizações e exclusões de dados em tabelas sink	Suportado

Pré-requisitos

Antes de começar, verifique se você possui:

Uma instância Tair (compatível com Redis OSS). Consulte Step 1: Crie an instance.
Uma lista de permissões configurada para a instância. Consulte Step 2: Configure whitelists.

Limitações

Semântica de entrega: O conector suporta apenas entrega com melhor esforço. Não há suporte à semântica exactly-once. Garanta que suas operações de gravação sejam idempotentes.
Tipos de dados de tabela de dimensão: Tabelas de dimensão leem apenas dados STRING e HASHMAP. Todos os campos devem ser do tipo STRING.
Chave primária da tabela de dimensão: Cada tabela de dimensão deve ter exatamente uma chave primária. A cláusula ON em um JOIN de tabela de dimensão deve usar condições de igualdade na chave primária.

Problemas conhecidos

VVR 8.0.9 — Bug de cache do Buffered Writer: A versão VVR 8.0.9 apresenta um problema de cache do Buffered Writer. Como solução alternativa, defina sink.buffer-flush.max-rows como 0 na cláusula WITH da tabela sink.

Sintaxe

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

Opções do conector

Opções gerais

Estas opções aplicam-se tanto a tabelas sink quanto a tabelas de dimensão.

Opção	Tipo de dados	Obrigatório	Padrão	Descrição
`connector`	STRING	Sim	—	Defina como `redis`.
`host`	STRING	Sim	—	Endereço IP para conexão ao banco de dados ApsaraDB for Redis. Sempre que possível, utilize o endpoint interno. Conexões via internet podem apresentar maior latência ou limitações de largura de banda.
`port`	INT	Não	`6379`	Número da porta.
`password`	STRING	Não	(string vazia)	Senha de acesso.
`dbNum`	INT	Não	`0`	Número sequencial do banco de dados.
`clusterMode`	BOOLEAN	Não	`false`	Indica se o banco de dados está em modo cluster.
`hostAndPorts`	STRING	Não	—	Pares de host e porta no formato `"host1:port1,host2:port2"`. Obrigatório quando `clusterMode` é `true` e alta disponibilidade (HA) é necessária para conexões Jedis ao cluster Redis autogerenciado. Tem precedência sobre `host` e `port`. Se `clusterMode` for `true`, mas HA não for necessária, configure apenas `host` e `port` para especificar um único nó.
`key-prefix`	STRING	Não	—	Prefixo adicionado ao valor da chave primária ao ler de uma tabela de dimensão ou gravar em uma tabela sink. Um delimitador especificado por `key-prefix-delimiter` separa o prefixo e o valor da chave primária. Requer VVR 8.0.7 ou posterior.
`key-prefix-delimiter`	STRING	Não	—	Delimitador entre o prefixo da chave e o valor da chave primária.
`connection.pool.max-total`	INT	Não	`8`	Número máximo de conexões que o pool pode alocar. Requer VVR 8.0.9 ou posterior.
`connection.pool.max-idle`	INT	Não	`8`	Número máximo de conexões ociosas no pool.
`connection.pool.min-idle`	INT	Não	`0`	Número mínimo de conexões ociosas no pool.
`connection.pool.lifo`	Boolean	Não	`true`	Define se as conexões ociosas são alocadas do pool em ordem LIFO. Valores válidos: true: LIFO. O pool aloca primeiro a conexão devolvida mais recentemente. false: FIFO. A conexão ociosa menos usada recentemente é alocada primeiro. Para uma instância de cluster Redis atrás de um proxy, definir isso como false pode balancear a carga entre os proxies de forma mais uniforme e evitar desvio de conexões. No entanto, FIFO tende a manter as conexões ocupadas e pode aumentar o número total de conexões; avalie isso cuidadosamente em cenários com muitas conexões. Nota: Esta opção é suportada apenas no mecanismo Realtime Compute VVR 11.8.0 e posterior.
`connect.timeout`	DURATION	Não	`3000ms`	Tempo limite para estabelecimento de conexão.
`socket.timeout`	DURATION	Não	`3000ms`	Tempo limite para recebimento de dados do servidor Redis.
`cacert.filepath`	STRING	Não	—	Caminho completo para o certificado SSL/TLS. O arquivo deve estar no formato JKS. Se não definido, a criptografia SSL/TLS fica desativada. Para ativar a criptografia, baixe o certificado CA e faça upload dele como dependência adicional — ele é armazenado no diretório `/flink/usrlib`. Exemplo: `'cacert.filepath' = '/flink/usrlib/ca.jks'`. Requer VVR 11.1 ou posterior.

Opções de tabela sink

Opção	Tipo de dados	Obrigatório	Padrão	Descrição
`mode`	STRING	Sim	—	Estrutura de dados Redis para a tabela sink. Cinco estruturas são suportadas: STRING, LIST, SET, HASHMAP e SORTEDSET. A instrução DDL deve corresponder à estrutura escolhida. Consulte Estruturas de dados para tabelas sink.
`flattenHash`	BOOLEAN	Não	`false`	Define se os dados HASHMAP são gravados no modo multivalor. Quando `true`, declare vários campos que não são chave primária: a chave primária mapeia para a chave Redis, cada nome de campo que não é chave primária mapeia para um campo Hash, e cada valor de campo mapeia para o valor Hash. Quando `false` (modo valor único), declare exatamente três campos: a chave primária mapeia para a chave, o primeiro campo que não é chave primária mapeia para o campo Hash, e o segundo mapeia para o valor Hash. Tem efeito apenas quando `mode` é `HASHMAP`. Requer VVR 8.0.7 ou posterior.
`ignoreDelete`	BOOLEAN	Não	`false`	Define se as mensagens de retração são ignoradas. Quando `true`, as mensagens de retração são descartadas. Quando `false`, a chave e seus dados são excluídos ao receber uma mensagem de retração.
`expiration`	LONG	Não	`0`	Tempo de vida (TTL) das chaves inseridas, em milissegundos. `0` desativa o TTL.
`sink.buffer-flush.max-rows`	INT	Não	`200`	Número máximo de registros (eventos de adição, modificação e exclusão) mantidos no buffer antes do flush. Requer VVR 8.0.9 ou posterior para `clusterMode = false`; VVR 11.4.0 ou posterior para `clusterMode = true`.
`sink.buffer-flush.interval`	DURATION	Não	`1000ms`	Intervalo no qual o buffer realiza flush assíncrono. Requer VVR 8.0.9 ou posterior para `clusterMode = false`; VVR 11.4.0 ou posterior para `clusterMode = true`.

Opções de tabela de dimensão

Opção	Tipo de dados	Obrigatório	Padrão	Descrição
`mode`	STRING	Não	`STRING`	Tipo de dado a ser lido da tabela de dimensão. `STRING` lê dados STRING. `HASHMAP` lê dados Hash aninhados (Key → Map\): declare várias colunas que não são chave primária, onde a chave primária mapeia para a chave Redis, cada nome de coluna que não é chave primária mapeia para um campo Hash, e seu valor mapeia para o valor do campo. Requer VVR 8.0.7 ou posterior. Para ler dados HASHMAP no modo valor único, defina `hashName` em vez disso.
`hashName`	STRING	Não	—	Chave Hash fixa usada ao ler dados HASHMAP no modo valor único. Quando definido, declare dois campos: a chave primária mapeia para o campo Hash, e a chave que não é primária mapeia para o valor Hash.
`cache`	STRING	Não	`None`	Política de cache. `None` desativa o cache. `LRU` armazena em cache um subconjunto de dados — em caso de falha no cache, o conector consulta a tabela de dimensão. `ALL` carrega toda a tabela de dimensão no cache antes da execução do deployment; todas as consultas subsequentes usam o cache, e o cache é recarregado após a expiração das entradas. Consulte Notas de uso para a opção cache.
`cacheSize`	LONG	Não	`10000`	Número máximo de linhas a serem armazenadas em cache. Obrigatório quando `cache` é `LRU`.
`cacheTTLMs`	LONG	Não	—	Tempo limite do cache em milissegundos. Para `LRU`, define a expiração por entrada (sem expiração por padrão). Para `ALL`, define o intervalo de recarga do cache (sem recarga por padrão). Não tem efeito quando `cache` é `None`.
`cacheEmpty`	BOOLEAN	Não	`true`	Define se resultados vazios (sem correspondência) são armazenados em cache.
`cacheReloadTimeBlackList`	STRING	Não	—	Períodos durante os quais a política de cache `ALL` não recarrega. Útil durante eventos de alto tráfego. Formato: use `->` para separar horários de início e fim, e `,` para separar múltiplos períodos. Exemplos: dia único `2017-10-24 14:00 -> 2017-10-24 15:00`; entre dias `2017-11-10 23:30 -> 2017-11-11 08:00`; recorrência diária `12:00 -> 14:00, 22:00 -> 2:00` (requer VVR 11.1 ou posterior).
`async`	BOOLEAN	Não	`false`	Define se a consulta assíncrona é ativada. Quando `true`, os resultados são retornados fora de ordem.

Notas de uso para a opção cache

ALL requer VVR 8.0.3 ou posterior.
Para versões de VVR 8.0.3 até VVR 11.1 (exclusive), cache = ALL lê HASHMAP apenas no modo valor único. Na cláusula WITH do DDL, defina hashName como o nome da chave; declare Field como chave primária e Value como coluna que não é chave primária.
A partir do VVR 11.1, cache = ALL suporta o modo multivalor HASHMAP. Especifique a chave Redis como chave primária e declare várias colunas que não são chave primária para cada campo Hash. Defina mode = HASHMAP na cláusula WITH.
A opção cache deve ser usada em conjunto com cacheSize e cacheTTLMs.

Estruturas de dados para tabelas sink

Cada estrutura de dados Redis corresponde a um esquema DDL e comando de gravação específicos.

Estrutura de dados	Esquema DDL	Comando de gravação
STRING	Duas colunas: key (STRING), value (STRING)	`set key value`
LIST	Duas colunas: key (STRING), value (STRING)	`lpush key value`
SET	Duas colunas: key (STRING), value (STRING)	`sadd key value`
HASHMAP (modo valor único, padrão)	Três colunas: key (STRING), field (STRING), value (STRING)	`hmset key field value`
HASHMAP (modo multivalor, `flattenHash = true`)	Múltiplas colunas: key (STRING), seguida de uma coluna por campo Hash — cada nome de coluna é o nome do campo e seu valor é o valor do campo	`hmset key col1 value1 col2 value2 ...`
SORTEDSET	Três colunas: key (STRING), score (DOUBLE), value (STRING)	`zadd key score value`

A opção ignoreDelete controla como as mensagens de retração são tratadas. Quando definida como true , as operações de exclusão são ignoradas.

Mapeamentos de tipos de dados

Escopo	Tipo Tair (ApsaraDB for Redis)	Tipo Flink
Todos os tipos de tabela	STRING	STRING
Apenas tabelas sink	SCORE	DOUBLE

O tipo SCORE é usado com dados SORTEDSET. Cada valor em um conjunto ordenado requer uma pontuação DOUBLE, e os valores são classificados em ordem crescente de suas pontuações.

Exemplos

Exemplos de tabela sink

Todos os exemplos de tabela sink leem de uma source Kafka e gravam em uma sink Tair.

Gravar dados STRING

Este exemplo usa user_id como chave Redis e login_time como valor 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;

Gravar dados HASHMAP no modo multivalor

Este exemplo usa order_id como chave Redis e grava product_name, quantity e amount como campos Hash separados.

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;

Gravar dados HASHMAP no modo valor único

Este exemplo usa order_id como chave Redis, product_name como campo Hash e 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;

Exemplos de tabela de dimensão

Todos os exemplos de tabela de dimensão consultam informações de usuário em uma tabela de dimensão Tair e as unem com um stream Kafka.

Ler dados STRING

Este exemplo usa user_id como chave Redis e recupera user_name como valor 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;

Ler dados HASHMAP no modo multivalor

Este exemplo usa user_id como chave Redis e recupera múltiplos campos Hash — user_name, email e register_time — em uma única consulta.

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;

Ler dados HASHMAP no modo valor único

Este exemplo usa uma chave Hash fixa (testkey) definida via hashName. A coluna user_id mapeia para o campo Hash, e user_name mapeia para o 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;