Realtime Compute for Apache Flink:Upsert Kafka

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

• Un cluster Kafka. Créez un cluster DataFlow Kafka ou créez des ressources dans ApsaraMQ for Kafka

• Une connectivité réseau entre votre cluster Flink et votre cluster Kafka. Pour Kafka sur EMR, configurez un VPC et un groupe de sécurité. Pour ApsaraMQ for Kafka, configurez une liste d'autorisation

Fonctionnement

Le flux de journal des modifications upsert produit par la source contient uniquement des entrées +I (insertion), +U (mise à jour après) et -D (suppression). Contrairement à un flux retract, il ne contient jamais de messages -U (mise à jour avant).

En tant que table source, le connecteur lit les enregistrements Kafka et les convertit en un flux de journal des modifications :

• Un enregistrement dont la clé existe déjà est interprété comme une MISE À JOUR (+U)

• Un enregistrement dont la clé n'existe pas est interprété comme une INSERTION (+I)

• Un enregistrement avec une valeur nulle est interprété comme une SUPPRESSION (-D)

En tant que table puits ou puits d'ingestion de données, le connecteur consomme un flux de journal des modifications en amont :

• Les enregistrements INSERT et UPDATE_AFTER sont écrits sous forme de messages Kafka normaux

• Les enregistrements DELETE sont écrits sous forme de messages tombstone (valeur nulle pour la clé correspondante)

Flink partitionne la sortie du puits par clé primaire, de sorte que tous les messages pour la même clé arrivent dans la même partition dans le bon ordre.

Exemple

L'exemple de bout en bout suivant lit les événements de consultation de pages web depuis une table source Kafka, agrège les vues de page (PV) et les visiteurs uniques (UV) par région, et écrit les résultats dans un puits Upsert Kafka.

Table source — lit les événements bruts de consultation de pages :

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

Table puits — stocke les résultats agrégés avec la sémantique 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'
);

Chaque table Upsert Kafka doit définir une clé primaire. Les colonnes de la clé primaire déterminent comment Flink partitionne la sortie et comment le connecteur identifie les opérations upsert.

Requête — agrège et écrit les résultats :

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

Ingestion de données

Le connecteur Upsert Kafka peut servir de puits dans un job YAML d'ingestion de données. Les données sont écrites au format JSON, et les champs de clé primaire sont inclus dans le corps du message.

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}

Sémantique de livraison

Par défaut, le puits Upsert Kafka écrit avec des garanties au moins une fois (at-least-once). Flink peut écrire des enregistrements en double avec la même clé dans la rubrique. Comme le connecteur fonctionne en mode upsert, le dernier enregistrement pour une clé donnée prend effet lors de la relecture de la rubrique en tant que source. Les écritures en double sont donc idempotentes — elles ne corrompent pas l'état final.

Pour activer la sémantique exactement une fois (exactly-once), le cluster Kafka doit prendre en charge les transactions Kafka (Apache Kafka 0.11 ou version ultérieure) et la fonctionnalité de transaction doit être activée. Définissez sink.delivery-guarantee sur exactly-once dans la configuration YAML d'ingestion de données.

Limites

• Nécessite Flink exécuté sur Ververica Runtime (VVR) 2.0.0 ou version ultérieure.

• Prend en charge Apache Kafka 0.10 ou version ultérieure pour les lectures et les écritures.

• Prend en charge uniquement les paramètres client documentés pour Apache Kafka 2.8. Consultez les références de configuration consommateur et producteur.

• Les tables source Upsert Kafka commencent toujours à partir de earliest-offset — ce paramètre est fixe et non configurable. Le connecteur doit lire toutes les données de modification historiques pour reconstruire un journal des modifications complet. L'utilisation de tout autre mode de démarrage (tel que latest-offset ou par horodatage) entraînerait un journal des modifications incomplet et provoquerait des problèmes d'exactitude des données dans les calculs en aval.

• Pour la sémantique de puits exactly-once, le cluster Kafka de destination doit exécuter Apache Kafka 0.11 ou version ultérieure avec les transactions activées.

Métriques de surveillance

Métriques de la table source

Métriques de la table puits

Étapes suivantes

• Synchroniser une base de données MySQL entière vers Kafka à l'aide de Flink CDC

• Résoudre les conflits de dépendances Flink