Realtime Compute for Apache Flink:Kafka connector

Visão geral

Pré-requisitos

Observações

Gravações transacionais não são recomendadas devido a limitações de design conhecidas no Apache Flink e no Apache Kafka. Ao definir sink.delivery-guarantee = 'exactly-once', o Kafka connector ativa as gravações transacionais, com os seguintes problemas conhecidos:

• Cada checkpoint gera um novo Transaction ID. Caso o intervalo entre checkpoints seja muito curto, o grande volume de Transaction IDs gerados pode esgotar a memória do coordenador do cluster Kafka, comprometendo sua estabilidade.

• Cada transação cria uma nova instância de Producer. Se muitas transações forem confirmadas simultaneamente, o TaskManager pode ficar sem memória, desestabilizando o job do Apache Flink.

• Se múltiplos jobs do Apache Flink utilizarem o mesmo sink.transactional-id-prefix, os Transaction IDs gerados poderão entrar em conflito. Quando uma operação de gravação falha em um job, isso pode impedir o avanço do Log Start Offset (LSO) de uma partição do Apache Kafka, afetando todos os consumidores daquela partição.

Caso necessite de semântica de exactly-once, use o Upsert Kafka connector para gravar em uma tabela de chave primária, garantindo idempotência. Se o uso de gravações transacionais for indispensável, consulte Exactly-once semantics usage notes.

Solucionar problemas de conectividade de rede

O erro Timed out waiting for a node assignment quando um job do Realtime Compute for Apache Flink falha ao iniciar geralmente indica um problema de conectividade de rede entre o Realtime Compute for Apache Flink e o cluster Kafka.

SQL

Use o Kafka connector como tabela de origem ou tabela de destino em jobs SQL.

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

Colunas de metadados

CREATE TABLE kafka_source (
  -- Read the message topic as the `record_topic` column
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- Read the timestamp from the ConsumerRecord as the `ts` column
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- Read the message offset as the `record_offset` column
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE kafka_sink (
  -- Write the timestamp from the `ts` column as the ProducerRecord's timestamp to Kafka
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

Opções do conector

• General

  Option	Description	Type	Required	Default	Remarks
  connector	Tipo do conector.	String	Yes	–	O valor deve ser kafka.
  properties.bootstrap.servers	Lista de endereços dos brokers Kafka.	String	Yes	–	Formato: host1:port1,host2:port2,.... Separe os endereços com vírgulas (,).
  properties.*	Propriedades adicionais para o cliente Kafka.	String	No	–	As chaves de propriedade devem ser opções válidas definidas na documentação oficial do Apache Kafka para Producer Configs e Consumer Configs. O Realtime Compute for Apache Flink remove o prefixo properties. e repassa os pares chave-valor restantes ao cliente Kafka subjacente. Por exemplo, defina 'properties.allow.auto.create.topics' = 'false' para desativar a criação automática de tópicos. O conector Kafka sobrescreve as opções a seguir, portanto não é possível configurá-las por este meio: key.deserializer value.deserializer
  format	Formato para serialização e desserialização do valor de uma mensagem Kafka.	String	No	–	Formatos suportados: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Nota Para mais informações, consulte Opções de formato.
  key.format	Formato para serialização e desserialização da chave de uma mensagem Kafka.	String	No	–	Formatos suportados: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Nota Ao usar esta configuração, key.options é obrigatório.
  key.fields	Campos do schema da tabela a serem usados como chave da mensagem Kafka.	String	No	–	Separe múltiplos nomes de campo com ponto e vírgula (;). Por exemplo, 'field1;field2'.
  key.fields-prefix	Prefixo personalizado para todos os campos de chave, evitando conflitos de nomes com os campos de valor.	String	No	–	Este prefixo distingue os campos de chave dos campos de valor. Ele é removido antes da serialização da chave ou após a desserialização. Nota Ao usar esta opção, value.fields-include deve ser definido como EXCEPT_KEY.
  value.format	Formato para serialização e desserialização do valor de uma mensagem Kafka.	String	No	–	Esta configuração equivale a format. Defina apenas uma das opções: format ou value.format. Se ambas forem configuradas, value.format tem precedência sobre format.
  value.fields-include	Define se os campos de chave são incluídos no formato de valor.	String	No	ALL	Valores válidos: ALL: O valor da mensagem Kafka inclui todas as colunas da tabela. EXCEPT_KEY: O valor da mensagem Kafka inclui todas as colunas da tabela, exceto as definidas em key.fields.

• Source table

  Option	Description	Type	Required	Default	Remarks
  topic	Tópico ou tópicos a serem lidos.	String	No	–	Para assinar múltiplos tópicos, separe os nomes com ponto e vírgula (;), por exemplo, 'topic-1;topic-2'. Nota Especifique esta opção ou topic-pattern, mas não ambas.
  topic-pattern	Expressão regular que corresponde aos tópicos a serem assinados. O consumidor assina todos os tópicos cujos nomes correspondam a este padrão.	String	No	–	Exemplos: user_event_.*: Corresponde a todos os tópicos com o prefixo user_event_. prod\.logs\..*: Corresponde a tópicos com o prefixo prod.logs. (o caractere . deve ser escapado). Nota Especifique esta opção ou topic, mas não ambas.
  properties.group.id	ID do grupo de consumidores da source Kafka.	String	No	KafkaSource-{Source-Table-Name}	Ao usar um ID de grupo de consumidores pela primeira vez, defina também properties.auto.offset.reset como earliest ou latest para estabelecer o offset inicial de leitura.
  scan.startup.mode	Offset de inicialização do consumidor Kafka.	String	No	group-offsets	Valores válidos: earliest-offset: Inicia a leitura a partir do offset mais antigo disponível. latest-offset: Inicia a leitura a partir do offset mais recente. group-offsets: Inicia a leitura a partir dos offsets confirmados do properties.group.id especificado. timestamp: Inicia a leitura a partir do valor especificado em scan.startup.timestamp-millis. specific-offsets: Inicia a leitura a partir dos offsets definidos em scan.startup.specific-offsets. Nota Esta opção aplica-se apenas quando um job é iniciado sem estado. Se um job retomar a partir de um checkpoint, a leitura começa pelos offsets armazenados no estado do checkpoint.
  scan.startup.specific-offsets	Offset de início por partição quando scan.startup.mode é specific-offsets.	String	No	–	Por exemplo, partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	Timestamp de início em milissegundos quando scan.startup.mode está definido como timestamp.	Long	No	–	A unidade é milissegundos.
  scan.topic-partition-discovery.interval	Intervalo de descoberta de partições.	Duration	No	5 minutes	O conector descobre e lê periodicamente novas partições. Ao usar topic-pattern, o conector também detecta novos tópicos que correspondam ao padrão. Defina o intervalo como um valor não positivo para desativar este recurso. Nota No Ververica Runtime (VVR) 6.0.x, a descoberta dinâmica de partições está desativada por padrão. A partir do VVR 8.0, este recurso é ativado por padrão com um intervalo de descoberta de 5 minutos.
  scan.header-filter	Filtra mensagens com base nos cabeçalhos da mensagem Kafka.	String	No	–	A chave e o valor do cabeçalho são separados por dois-pontos (:). Múltiplas condições de cabeçalho são conectadas por operadores lógicos (& e |). O operador lógico NOT (!) também é suportado. Por exemplo, depart:toy|depart:book&!env:test retém os dados Kafka se o cabeçalho contiver depart=toy ou depart=book e não contiver env=test. Nota Esta opção é suportada apenas no Ververica Runtime (VVR) 8.0.6 e versões posteriores. Parênteses em expressões não são suportados. As operações lógicas são avaliadas da esquerda para a direita. Os valores dos cabeçalhos são convertidos para strings UTF-8 para comparação.
  scan.check.duplicated.group.id	Verifica se outro consumidor ativo já está usando o properties.group.id.	Boolean	No	false	Valores válidos: true: Antes de iniciar o job, o sistema verifica a existência de um grupo de consumidores duplicado. Se encontrado, o job falha para evitar conflitos. false: Inicia o job sem verificar conflitos. Nota Esta opção é suportada apenas no Ververica Runtime (VVR) 6.0.4 e versões posteriores.

• Sink table

  Option	Description	Type	Required	Default	Remarks
  topic	Tópico de destino.	String	Yes	–	–
  sink.partitioner	Mapeia registros de instâncias paralelas do sink para partições Kafka.	String	No	default	Valores válidos: default: Usa o particionador padrão do Kafka. ### **Confirmação de offsets do source** A tabela source do Kafka confirma o offset do consumidor no cluster Kafka somente após um checkpoint bem-sucedido. Por isso, intervalos longos de checkpoint fazem com que o offset confirmado fique defasado. O progresso real de leitura fica armazenado no estado do checkpoint, que o sistema utiliza para recuperação de falhas. Os offsets confirmados servem apenas como monitor de progresso e não são usados na recuperação — portanto, falhas de confirmação não afetam a precisão dos dados. ### **Partitioner customizado para sink** Se a estratégia de particionamento nativa do Kafka não atender às suas necessidades, implemente um partitioner customizado estendendo a classe [FlinkKafkaPartitioner](https://github.com/apache/flink-connector-kafka/blob/369e7be46a70fd50d68746498aed82105741e7d6/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/FlinkKafkaPartitioner.java). Após o desenvolvimento, compile o código em um pacote JAR e faça o upload usando o recurso [Artifacts](t2740613.xdita#) no console do Realtime Compute. Com o pacote JAR carregado e referenciado, defina o parâmetro sink.partitioner na cláusula WITH com o nome completo e qualificado da sua classe de partitioner, por exemplo, `org.mycompany.MyPartitioner`. ### **Kafka, Upsert Kafka e Kafka JSON catalog** O Kafka é uma plataforma de streaming de eventos append-only que não suporta atualizações nem exclusões de dados. Em SQL de streaming, uma tabela sink Kafka padrão não consegue lidar com dados de Change Data Capture (CDC) provenientes de upstream nem com a lógica de retração de operadores como aggregate e join. Quando for necessário gravar dados que contenham alterações ou retrações, utilize uma tabela sink [Upsert Kafka](t2306784.dita#concept_2306784). Para simplificar a sincronização em lote de dados de Change Data Capture (CDC) de uma ou mais tabelas de banco de dados upstream para o Kafka, use um Kafka JSON catalog. Se os dados armazenados no Kafka estiverem em formato JSON, o Kafka JSON catalog permite pular a etapa de definição de schema e dos parâmetros WITH. Consulte [Manage Kafka JSON catalogs](t2212019.dita#task_2159375) para mais detalhes. ### **Exemplos** #### **Exemplo 1: Leitura e escrita no Kafka** Este exemplo lê dados de um tópico Kafka de source e os grava em um tópico de sink. Os dados estão no formato CSV. ```sql CREATE TEMPORARY TABLE kafka_source ( id INT, name STRING, age INT ) WITH ( 'connector' = 'kafka', 'topic' = 'source', 'properties.bootstrap.servers' = '', 'properties.group.id' = '', 'format' = 'csv' ); CREATE TEMPORARY TABLE kafka_sink ( id INT, name STRING, age INT ) WITH ( 'connector' = 'kafka', 'topic' = 'sink', 'properties.bootstrap.servers' = '', 'properties.group.id' = '', 'format' = 'csv' ); INSERT INTO kafka_sink SELECT id, name, age FROM kafka_source; ``` #### **Exemplo 2: Sincronização de schema e dados de tabela** Use o conector Kafka para sincronizar mensagens de um tópico Kafka para o Hologres em tempo real. Para evitar mensagens duplicadas no Hologres durante um failover, utilize o offset e o ID de partição das mensagens Kafka como chave primária composta. ```sql CREATE TEMPORARY TABLE kafkaTable ( `offset` INT NOT NULL METADATA, `part` BIGINT NOT NULL METADATA FROM 'partition', PRIMARY KEY (`part`, `offset`) NOT ENFORCED ) WITH ( 'connector' = 'kafka', 'properties.bootstrap.servers' = '', 'topic' = 'kafka_evolution_demo', 'scan.startup.mode' = 'earliest-offset', 'format' = 'json', 'json.infer-schema.flatten-nested-columns.enable' = 'true' -- Optional. Flattens all nested columns. ); CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH ( 'connector' = 'hologres' ) AS TABLE vvp.`default`.kafkaTable; ``` #### **Exemplo 3: Sincronização de chaves e valores do Kafka** Se a chave de uma mensagem Kafka contiver informações relevantes, sincronize tanto a chave quanto o valor. ```sql CREATE TEMPORARY TABLE kafkaTable ( `key_id` INT NOT NULL, `val_name` VARCHAR(200) ) WITH ( 'connector' = 'kafka', 'properties.bootstrap.servers' = '', 'topic' = 'kafka_evolution_demo', 'scan.startup.mode' = 'earliest-offset', 'key.format' = 'json', 'value.format' = 'json', 'key.fields' = 'key_id', 'key.fields-prefix' = 'key_', 'value.fields-prefix' = 'val_', 'value.fields-include' = 'EXCEPT_KEY' ); CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`( WITH ( 'connector' = 'hologres' ) AS TABLE vvp.`default`.kafkaTable; ``` Nota As chaves de mensagens do Kafka não suportam Schema Evolution nem análise automática de tipos. Declare o schema manualmente. #### **Exemplo 4: Sincronização de dados com transformações** Ao sincronizar dados do Kafka para o Hologres, pode ser necessário aplicar transformações leves. ```sql CREATE TEMPORARY TABLE kafkaTable ( `distinct_id` INT NOT NULL, `properties` STRING, `timestamp` TIMESTAMP_LTZ METADATA, `date` AS CAST(`timestamp` AS DATE) ) WITH ( 'connector' = 'kafka', 'properties.bootstrap.servers' = '', 'topic' = 'kafka_evolution_demo', 'scan.startup.mode' = 'earliest-offset', 'key.format' = 'json', 'value.format' = 'json', 'key.fields' = 'key_id', 'key.fields-prefix' = 'key_' ); CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH ( 'connector' = 'hologres' ) AS TABLE vvp.`default`.kafkaTable ADD COLUMN `order_id` AS COALESCE(JSON_VALUE(`properties`, '$.order_id'), 'default'); --Use COALESCE to handle null values. ``` #### **Exemplo 5: Análise de JSON aninhado** A seguir, um exemplo de mensagem JSON: ```json { "id": 101, "name": "VVP", "properties": { "owner": "Alibaba Cloud", "engine": "Flink" } } ``` Para evitar o uso de funções como `JSON_VALUE(payload, '$.properties.owner')` para analisar campos, defina a estrutura diretamente no DDL do Source: ```sql CREATE TEMPORARY TABLE kafka_source ( id VARCHAR, `name` VARCHAR, properties ROW<`owner` STRING, engine STRING> ) WITH ( 'connector' = 'kafka', 'topic' = 'xxx', 'properties.bootstrap.servers' = 'xxx', 'scan.startup.mode' = 'earliest-offset', 'format' = 'json' ); ``` Com essa abordagem, o Flink analisa o JSON em campos estruturados durante a fase de leitura. As consultas SQL subsequentes podem referenciar diretamente `properties.owner` sem chamadas adicionais de funções, o que melhora o desempenho geral. ## Semântica exactly-once Configure o nível de isolamento do consumidor Todas as aplicações que consomem dados do Kafka devem definir a propriedade isolation.level: read_committed: Lê apenas dados confirmados. read_uncommitted (Padrão): Permite a leitura de dados não confirmados. O modo EXACTLY_ONCE depende de `read_committed`. Sem essa configuração, os consumidores podem enxergar dados não confirmados, comprometendo a consistência. Timeout de transação e perda de dados Ao recuperar de um checkpoint, o Realtime Compute for Apache Flink considera apenas as transações confirmadas antes do início daquele checkpoint. Se o intervalo entre a falha do job e sua reinicialização exceder o timeout de transação do Kafka, o Kafka aborta automaticamente a transação aberta, podendo resultar em perda de dados. O valor padrão de transaction.max.timeout.ms para um broker Kafka é de 15 minutos. Por padrão, o Flink Kafka Sink define o parâmetro transaction.timeout.ms como 1 hora. Aumente o valor de transaction.max.timeout.ms no broker para que seja maior ou igual à configuração definida no Flink. Pool de produtores e checkpoints concorrentes O modo EXACTLY_ONCE utiliza um pool de produtores Kafka de tamanho fixo. Cada checkpoint consome um produtor desse pool. Se o número de checkpoints concorrentes exceder o tamanho do pool, o job falhará. Dimensione o pool de produtores com base no número máximo de checkpoints concorrentes. Restrições para redução de paralelismo Se um job falhar antes de o primeiro checkpoint ser concluído, as informações do pool de produtores original serão perdidas na reinicialização. Portanto, não reduza o paralelismo do job antes que o primeiro checkpoint seja concluído. Caso a redução seja inevitável, o novo paralelismo não deve ser inferior a FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR. Transações bloqueiam leituras No modo read_committed, qualquer transação que não tenha sido confirmada ou abortada bloqueia as operações de leitura em todo o tópico. Por exemplo: A Transação 1 grava dados. A Transação 2 grava mais dados e é confirmada. Enquanto a Transação 1 permanecer aberta, os dados da Transação 2 — já confirmada — ficam invisíveis para os consumidores. Esse comportamento traz as seguintes implicações: Em operação normal, a latência de visibilidade dos dados é aproximadamente igual ao intervalo de checkpoint. Se um job falhar, qualquer tópico em que ele estivesse gravando ficará bloqueado para os consumidores até que o job seja reiniciado ou a transação expire. Em casos extremos, o próprio processo de expiração da transação também pode afetar as operações de leitura. ## FAQ How do I retrieve JSON data from Kafka using Realtime Compute for Apache Flink? Why does an Event Time-based Window produce no output for a ApsaraMQ for Kafka Source Table? How do I connect to a secured Kafka Cluster? How do I resolve field name conflicts? Workspaces and Namespaces