Realtime Compute for Apache Flink:Upsert Kafka

Pré-requisitos

Antes de começar, certifique-se de que você tem:

• Um cluster Kafka. Crie um DataFlow Kafka cluster ou create resources in ApsaraMQ for Kafka

• Conectividade de rede entre seu cluster Flink e o cluster Kafka. Para Kafka no EMR, configure um VPC e security group. Para ApsaraMQ for Kafka, configure a whitelist

Como funciona

O stream de changelog de upsert produzido pelo source contém apenas entradas +I (insert), +U (update after) e -D (delete). Diferentemente de um stream de retract, ele nunca contém mensagens -U (update before).

Como tabela source, o conector lê registros do Kafka e os converte em um stream de changelog:

• Um registro cuja chave já existe é interpretado como UPDATE (+U)

• Um registro cuja chave não existe é interpretado como INSERT (+I)

• Um registro com valor nulo é interpretado como DELETE (-D)

Como tabela sink ou sink de ingestão de dados, o conector consome um stream de changelog upstream:

• Registros INSERT e UPDATE_AFTER são gravados como mensagens normais do Kafka

• Registros DELETE são gravados como mensagens tombstone (valor nulo para a chave correspondente)

O Flink particiona a saída do sink por chave primária, garantindo que todas as mensagens da mesma chave cheguem à mesma partição na ordem correta.

Exemplo

O exemplo de ponta a ponta a seguir lê eventos de visualização de página de um site a partir de uma tabela source do Kafka, agrega visualizações de página (PV) e visitantes únicos (UV) por região e grava os resultados em um sink Upsert Kafka.

Tabela source — lê eventos brutos de visualização de página:

CREATE TABLE pageviews (
  user_id    BIGINT,
  page_id    BIGINT,
  viewtime   TIMESTAMP,
  user_region STRING,
  WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND
) WITH (
  'connector'                  = 'kafka',
  'topic'                      = '<yourTopicName>',
  'properties.bootstrap.servers' = '<host:port>',
  'format'                     = 'json'
);

Tabela sink — armazena resultados agregados com semântica de upsert:

CREATE TABLE pageviews_per_region (
  user_region STRING,
  pv          BIGINT,
  uv          BIGINT,
  PRIMARY KEY (user_region) NOT ENFORCED
) WITH (
  'connector'                  = 'upsert-kafka',
  'topic'                      = '<yourTopicName>',
  'properties.bootstrap.servers' = '<host:port>',
  'key.format'                 = 'avro',
  'value.format'               = 'avro'
);

Toda tabela Upsert Kafka deve definir uma chave primária. As colunas da chave primária determinam como o Flink particiona a saída e como o conector identifica operações de upsert.

Consulta — agrega e grava resultados:

INSERT INTO pageviews_per_region
SELECT
  user_region,
  COUNT(*),
  COUNT(DISTINCT user_id)
FROM pageviews
GROUP BY user_region;

SQL

CREATE TABLE upsert_kafka_sink (
  user_region STRING,
  pv          BIGINT,
  uv          BIGINT,
  PRIMARY KEY (user_region) NOT ENFORCED
) WITH (
  'connector'                  = 'upsert-kafka',
  'topic'                      = '<yourTopicName>',
  'properties.bootstrap.servers' = '...',
  'key.format'                 = 'avro',
  'value.format'               = 'avro'
);

Ingestão de dados

O conector Upsert Kafka pode servir como sink em um job YAML de ingestão de dados. Os dados são gravados no formato JSON e os campos de chave primária são incluídos no corpo da mensagem.

sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: localhost:9092
  # ApsaraMQ for Kafka
  aliyun.kafka.accessKeyId: ${secret_values.kafka-ak}
  aliyun.kafka.accessKeySecret: ${secret_values.kafka-sk}
  aliyun.kafka.instanceId: ${instancd-id}
  aliyun.kafka.endpoint: ${endpoint}
  aliyun.kafka.regionId: ${region-id}

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604

sink:
  type: upsert-kafka
  name: Upsert Kafka Sink
  properties.bootstrap.servers: ${upsert.kafka.bootstraps.server}
  aliyun.kafka.accessKeyId: ${upsert.kafka.aliyun.ak}
  aliyun.kafka.accessKeySecret: ${upsert.kafka.aliyun.sk}
  aliyun.kafka.instanceId: ${upsert.kafka.aliyun.instanceid}
  aliyun.kafka.endpoint: ${upsert.kafka.aliyun.endpoint}
  aliyun.kafka.regionId: ${upsert.kafka.aliyun.regionid}

route:
  - source-table: ${mysql.source.table}
    sink-table: ${upsert.kafka.topic}

Semântica de entrega

Por padrão, o sink Upsert Kafka grava com garantias at-least-once. O Flink pode gravar registros duplicados com a mesma chave no tópico. Como o conector opera no modo upsert, o último registro de qualquer chave específica prevalece ao ler o tópico novamente como source. Gravações duplicadas são, portanto, idempotentes — elas não corrompem o estado final.

Para ativar a semântica exactly-once, o cluster Kafka deve suportar transações do Kafka (Apache Kafka 0.11 ou posterior) e o recurso de transação deve estar habilitado. Defina sink.delivery-guarantee como exactly-once na configuração YAML de ingestão de dados.

Limites

• Requer Flink executando no Ververica Runtime (VVR) 2.0.0 ou posterior.

• Suporta Apache Kafka 0.10 ou posterior tanto para leitura quanto para gravação.

• Suporta apenas os parâmetros de cliente documentados para o Apache Kafka 2.8. Consulte as referências de configuração de consumer e producer.

• Tabelas source Upsert Kafka sempre iniciam a partir de earliest-offset — isso é fixo e não configurável. O conector precisa ler todos os dados históricos de alterações para reconstruir um changelog completo. Usar qualquer outro modo de inicialização (como latest-offset ou por timestamp) resultaria em um changelog incompleto e causaria problemas de integridade dos dados nos cálculos downstream.

• Para semântica exactly-once no sink, o cluster Kafka de destino deve executar o Apache Kafka 0.11 ou posterior com transações habilitadas.

Métricas de monitoramento

Métricas da tabela source

Métricas da tabela sink

Próximos passos

• Synchronize an entire MySQL database to Kafka using Flink CDC

• Resolve Flink dependency conflicts