Realtime Compute for Apache Flink:Kafka connector

Présentation

Prérequis

Remarques

Les écritures transactionnelles ne sont pas recommandées en raison de limitations de conception connues dans Apache Flink et Apache Kafka. Lorsque vous définissez sink.delivery-guarantee = 'exactly-once', le connecteur Kafka active les écritures transactionnelles, avec les problèmes connus suivants :

• Chaque point de contrôle génère un nouvel ID de transaction. Si l'intervalle entre les points de contrôle est trop court, le nombre élevé d'ID de transaction peut saturer la mémoire du coordinateur du cluster Kafka, compromettant ainsi la stabilité du cluster.

• Chaque transaction crée une nouvelle instance de producteur. Si trop de transactions sont validées simultanément, le TaskManager peut manquer de mémoire, déstabilisant ainsi la tâche Apache Flink.

• Si plusieurs tâches Apache Flink utilisent le même préfixe sink.transactional-id-prefix, les ID de transaction générés peuvent entrer en conflit. Lorsqu'une opération d'écriture échoue dans une tâche, cela peut empêcher l'avancement du Log Start Offset (LSO) d'une partition Apache Kafka. Cela affecte tous les consommateurs de cette partition.

Si vous avez besoin de la sémantique exactly-once, utilisez le connecteur Upsert Kafka pour écrire dans une table avec clé primaire, garantissant ainsi l'idempotence. Si vous devez utiliser des écritures transactionnelles, consultez les remarques d'utilisation de la sémantique exactly-once.

Dépannage de la connectivité réseau

Une erreur Timed out waiting for a node assignment lors du démarrage d'une tâche Realtime Compute for Apache Flink indique généralement un problème de connectivité réseau entre Realtime Compute for Apache Flink et le cluster Kafka.

SQL

Utilisez le connecteur Kafka en tant que table source ou table de destination dans les tâches 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'
)

Colonnes de métadonnées

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

Options du connecteur

• Général

  Option	Description	Type	Required	Default	Remarks
  connector	Type de connecteur.	String	Yes	–	La valeur doit être kafka.
  properties.bootstrap.servers	Liste des adresses des brokers Kafka.	String	Yes	–	Format : host1:port1,host2:port2,.... Séparez les adresses par des virgules (,).
  properties.*	Propriétés supplémentaires pour le client Kafka.	String	No	–	Les clés de propriété doivent correspondre aux options valides définies dans la documentation officielle d'Apache Kafka pour les configurations du producteur et les configurations du consommateur. Realtime Compute for Apache Flink supprime le préfixe properties. et transmet les paires clé-valeur restantes au client Kafka sous-jacent. Par exemple, vous pouvez définir 'properties.allow.auto.create.topics' = 'false' pour désactiver la création automatique de topics. Le connecteur Kafka écrase ces options ; vous ne pouvez donc pas les configurer de cette manière : key.deserializer value.deserializer
  format	Format de sérialisation et de désérialisation de la valeur d'un message Kafka.	String	No	–	Formats pris en charge : csv json avro debezium-json canal-json maxwell-json avro-confluent raw Remarque Pour plus d'informations, consultez les options de format.
  key.format	Format de sérialisation et de désérialisation de la clé d'un message Kafka.	String	No	–	Formats pris en charge : csv json avro debezium-json canal-json maxwell-json avro-confluent raw Remarque Lorsque vous utilisez cette configuration, key.options est obligatoire.
  key.fields	Champs du schéma de table à utiliser comme clé de message Kafka.	String	No	–	Séparez plusieurs noms de champs par des points-virgules (;). Par exemple, 'field1;field2'.
  key.fields-prefix	Préfixe personnalisé pour tous les champs de clé afin d'éviter les conflits de nom avec les champs de valeur.	String	No	–	Ce préfixe permet de distinguer les champs de clé des champs de valeur. Il est supprimé avant la sérialisation de la clé ou après sa désérialisation. Remarque Si vous utilisez cette option, value.fields-include doit être défini sur EXCEPT_KEY.
  value.format	Format de sérialisation et de désérialisation de la valeur d'un message Kafka.	String	No	–	Cette configuration équivaut à format. Vous ne pouvez définir que l'une des options format ou value.format. Si les deux sont configurées, value.format remplace format.
  value.fields-include	Indique si les champs de clé sont inclus dans le format de valeur.	String	No	ALL	Valeurs valides : ALL : la valeur du message Kafka inclut toutes les colonnes de la table. EXCEPT_KEY : la valeur du message Kafka inclut toutes les colonnes de la table, à l'exception de celles définies dans key.fields.

• Table source

  Option	Description	Type	Required	Default	Remarks
  topic	Topic ou topics à lire.	String	No	–	Pour s'abonner à plusieurs topics, séparez leurs noms par des points-virgules (;), par exemple 'topic-1;topic-2'. Remarque Vous pouvez spécifier cette option ou topic-pattern, mais pas les deux simultanément.
  topic-pattern	Expression régulière correspondant aux topics auxquels s'abonner. Le consommateur s'abonne à tous les topics dont les noms correspondent à ce modèle.	String	No	–	Exemples : user_event_.* : correspond à tous les topics préfixés par user_event_. prod\.logs\..* : correspond aux topics préfixés par prod.logs. (le caractère . doit être échappé). Remarque Vous pouvez spécifier cette option ou topic, mais pas les deux simultanément.
  properties.group.id	ID du groupe de consommateurs source Kafka.	String	No	KafkaSource-{Source-Table-Name}	Si vous utilisez un ID de groupe de consommateurs pour la première fois, vous devez également définir properties.auto.offset.reset sur earliest ou latest afin de définir le décalage de démarrage initial.
  scan.startup.mode	Décalage de démarrage du consommateur Kafka.	String	No	group-offsets	Valeurs valides : earliest-offset : commence la lecture à partir du décalage disponible le plus ancien. latest-offset : commence la lecture à partir du dernier décalage. group-offsets : commence la lecture à partir des décalages validés du properties.group.id spécifié. timestamp : commence la lecture à partir du scan.startup.timestamp-millis spécifié. specific-offsets : commence la lecture à partir des décalages spécifiés dans scan.startup.specific-offsets. Remarque Cette option s'applique uniquement lorsqu'un travail démarre sans état. Si un travail reprend à partir d'un point de contrôle, il lit à partir des décalages stockés dans l'état du point de contrôle.
  scan.startup.specific-offsets	Décalage de démarrage par partition lorsque scan.startup.mode est défini sur specific-offsets.	String	No	–	Par exemple, partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	Horodatage de démarrage en millisecondes lorsque scan.startup.mode est défini sur timestamp.	Long	No	–	L'unité est la milliseconde.
  scan.topic-partition-discovery.interval	Intervalle de découverte des partitions.	Duration	No	5 minutes	Le connecteur découvre et lit périodiquement les nouvelles partitions. Lorsque vous utilisez topic-pattern, le connecteur découvre également les nouveaux topics correspondant au modèle. Définissez l'intervalle sur une valeur non positive pour désactiver cette fonctionnalité. Remarque Dans Ververica Runtime (VVR) 6,0.x, la découverte dynamique des partitions est désactivée par défaut. À partir de VVR 8.0, cette fonctionnalité est activée par défaut avec un intervalle de découverte de 5 minutes.
  scan.header-filter	Filtre les messages en fonction des en-têtes de message Kafka.	String	No	–	Une clé d'en-tête et sa valeur sont séparées par deux-points (:). Plusieurs conditions d'en-tête sont reliées à l'aide d'opérateurs logiques (& et |). L'opérateur logique NON (!) est également pris en charge. Par exemple, depart:toy|depart:book&!env:test conserve les données Kafka si l'en-tête contient depart=toy ou depart=book et ne contient pas env=test. Remarque Cette option est prise en charge uniquement dans Ververica Runtime (VVR) 8.0.6 et versions ultérieures. Les parenthèses dans les expressions ne sont pas prises en charge. Les opérations logiques sont évaluées de gauche à droite. Les valeurs d'en-tête sont converties en chaînes UTF-8 pour la comparaison.
  scan.check.duplicated.group.id	Vérifie si un autre consommateur actif utilise déjà le properties.group.id.	Boolean	No	false	Valeurs valides : true : avant de démarrer le travail, le système vérifie l'existence d'un groupe de consommateurs dupliqué. Si un tel groupe est trouvé, le travail échoue pour éviter les conflits. false : démarre le travail sans vérifier les conflits. Remarque Cette option est prise en charge uniquement dans Ververica Runtime (VVR) 6.0.4 et versions ultérieures.

• Table sink

  Option	Description	Type	Required	Default	Remarks
  topic	Topic cible.	String	Yes	–	–
  sink.partitioner	Mappe les enregistrements des instances de sink parallèles vers les partitions Kafka.	String	No	default	Valeurs valides : default : utilise le partitionneur Kafka par défaut. fixed : chaque instance de sink parallèle écrit dans une partition Kafka fixe. round-robin : les enregistrements sont distribués aux partitions selon un algorithme tournant. Partitionneur personnalisé : pour utiliser un partitionneur personnalisé, indiquez le nom de classe qualifié complet d'une sous-classe FlinkKafkaPartitioner, par exemple org.mycompany.MyPartitioner.
  sink.delivery-guarantee	Garantie de livraison du sink.	String	No	at-least-once	Valeurs valides : none : aucune garantie n'est fournie. Les enregistrements peuvent être perdus ou dupliqués. at-least-once : garantit qu'aucun enregistrement n'est perdu, mais ils peuvent être dupliqués. exactly-once : utilise les transactions Kafka pour fournir une sémantique exactly-once, garantissant que les enregistrements ne sont ni perdus ni dupliqués. Remarque Lorsque vous utilisez la sémantique exactly-once, vous devez également spécifier sink.transactional-id-prefix.
  sink.transactional-id-prefix	Préfixe d'ID de transaction. Obligatoire lorsque sink.delivery-guarantee est défini sur exactly-once.	String	Yes, if sink.delivery-guarantee is exactly-once	–	Requis uniquement lorsque sink.delivery-guarantee est défini sur exactly-once.
  sink.parallelism	Parallélisme de l'opérateur de sink.	Integer	No	–	Par défaut, le framework détermine le parallélisme en fonction des opérateurs en amont.

Sécurité et authentification

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";'
)

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_SSL',
  /* SSL configuration */
  /* Path to the truststore for the server's CA certificate. */
  /* Files uploaded using Artifacts are stored in the /flink/usrlib/ directory. */
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /* If client authentication is required, you must also configure the path to the keystore (private key). */
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /* The algorithm used to verify the server hostname. An empty string disables hostname verification. */
  'properties.ssl.endpoint.identification.algorithm' = '',
  /* SASL configuration */
  /* Set the SASL mechanism to SCRAM-SHA-256. */
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /* Configure JAAS. */
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

Vous pouvez utiliser la fonctionnalité Artifacts de la console Realtime Compute for Apache Flink pour télécharger le certificat CA et la clé privée mentionnés dans l'exemple. Les fichiers téléchargés sont stockés dans le répertoire /flink/usrlib. Pour utiliser un fichier de certificat CA nommé my-truststore.jks, vous pouvez définir la propriété 'properties.ssl.truststore.location' dans la clause WITH de l'une des deux manières suivantes :

• Définissez 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks'. Cette méthode évite de télécharger dynamiquement des fichiers depuis Object Storage Service (OSS) au moment de l'exécution, mais elle ne prend pas en charge le mode Debug.

• Si la version du moteur Realtime Compute est VVR 11.5 ou ultérieure, vous pouvez configurer properties.ssl.truststore.location et properties.ssl.keystore.location avec un chemin OSS absolu. Le format du chemin de fichier est oss://flink-fullymanaged-<Workspace ID>/artifacts/namespaces/<Namespace name>/<file name>. Cette méthode télécharge dynamiquement les fichiers OSS lors de l'exécution de Flink et prend en charge le mode Debug.

Offset de démarrage de la table source

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  -- Consume from the earliest offset.
  'scan.startup.mode' = 'earliest-offset',
  -- Consume from the latest offset.
  'scan.startup.mode' = 'latest-offset',
  -- Consume from the committed offsets of the consumer group "my-group".
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- If "my-group" is used for the first time, consumption starts from the earliest offset.
  'properties.auto.offset.reset' = 'latest', -- If "my-group" is used for the first time, consumption starts from the latest offset.
  -- Consume from the specified timestamp in milliseconds: 1655395200000.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Consume from specific offsets.
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

Validation des offsets source

La table source Kafka valide son offset de consommateur auprès du cluster Kafka uniquement après un point de contrôle (checkpoint) réussi ; ainsi, un intervalle de checkpoint long entraîne un décalage de l'offset validé. La table source stocke la progression réelle de la lecture dans l'état du checkpoint, que le système utilise pour la récupération après incident. Les offsets validés servent uniquement de moniteur de progression et ne sont pas utilisés pour la récupération ; par conséquent, les échecs de validation n'affectent pas l'exactitude des données.

Partitionneur de sink personnalisé

Si la stratégie de partitionnement intégrée de Kafka ne répond pas à vos besoins, vous pouvez implémenter un partitionneur personnalisé en étendant la classe FlinkKafkaPartitioner. Une fois le développement terminé, compilez votre code dans un package JAR et téléchargez-le à l'aide de la fonctionnalité Artifacts dans la console Realtime Compute. Après le téléchargement et la référence du package JAR, définissez le paramètre sink.partitioner dans la clause WITH avec le nom de classe complet de votre partitionneur, par exemple org.mycompany.MyPartitioner.

Kafka, Upsert Kafka et catalogue Kafka JSON

Kafka est une plateforme de streaming d'événements en ajout seul qui ne prend pas en charge les mises à jour ou les suppressions de données. En SQL streaming, une table sink Kafka standard ne peut pas gérer les données CDC (Change Data Capture) en amont ni la logique de rétractation des opérateurs tels que l'agrégation et la jointure. Si vous devez écrire des données contenant des modifications ou des rétractations, utilisez une table sink Upsert Kafka.

Pour simplifier la synchronisation par lot des données CDC (Change Data Capture) d'une ou plusieurs tables de base de données en amont vers Kafka, vous pouvez utiliser un catalogue Kafka JSON. Si les données stockées dans Kafka sont au format JSON, un catalogue Kafka JSON vous permet d'ignorer l'étape de définition du schéma et des paramètres WITH. Pour plus de détails, consultez Gérer les catalogues Kafka JSON.

Exemples

CREATE TEMPORARY TABLE kafka_source (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'source',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE kafka_sink (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

INSERT INTO kafka_sink SELECT id, name, age FROM kafka_source;

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' = '<yourKafkaBrokers>',
  '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;

CREATE TEMPORARY TABLE kafkaTable (
  `key_id` INT NOT NULL,
  `val_name` VARCHAR(200)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  '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;

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' = '<yourKafkaBrokers>',
  '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.

{
  "id": 101,
  "name": "VVP",
  "properties": {
    "owner": "Alibaba Cloud",
    "engine": "Flink"
  }
}

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

Sémantique Exactly-once

• Configurer le niveau d'isolation du consommateur

  Toutes les applications qui consomment des données Kafka doivent définir la propriété isolation.level :

  • read_committed : Lit uniquement les données validées.

  • read_uncommitted (Par défaut) : Peut lire des données non validées.

  EXACTLY_ONCE dépend de read_committed. Sinon, les consommateurs peuvent voir des données non validées, ce qui rompt la cohérence.

• Délai d'expiration de la transaction et perte de données

  Lors de la récupération à partir d'un checkpoint, Realtime Compute for Apache Flink ne prend en compte que les transactions qui ont été validées avant le début de ce checkpoint. Si la durée entre une panne du job et son redémarrage dépasse le délai d'expiration de la transaction Kafka, Kafka abandonne automatiquement la transaction ouverte, ce qui peut entraîner une perte de données.

  • Le transaction.max.timeout.ms par défaut pour un broker Kafka est de 15 minutes.

  • Par défaut, Flink Kafka Sink définit le paramètre transaction.timeout.ms sur 1 heure.

  • Vous devez augmenter transaction.max.timeout.ms sur le broker pour qu'il soit supérieur ou égal au paramètre défini dans Flink.

• Pool de producteurs et checkpoints concurrents

  Le mode EXACTLY_ONCE utilise un pool de producteurs Kafka de taille fixe. Chaque checkpoint utilise un producteur de ce pool. Si le nombre de checkpoints concurrents dépasse la taille du pool, le job échoue.

  Configurez la taille du pool de producteurs en fonction du nombre maximal de checkpoints concurrents.

• Contraintes de réduction du parallélisme

  Si un job échoue avant que le premier checkpoint ne soit terminé, les informations du pool de producteurs d'origine sont perdues au redémarrage. Par conséquent, ne réduisez pas le parallélisme du job avant que le premier checkpoint ne soit terminé. Si une réduction est nécessaire, le nouveau parallélisme ne doit pas être inférieur à FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR.

• Les transactions bloquent les lectures

  En mode read_committed, toute transaction qui n'a pas été validée ou abandonnée bloque les opérations de lecture sur l'ensemble du topic.

  Par exemple :

  • La transaction 1 écrit des données.

  • La transaction 2 écrit davantage de données et est validée.

  • Tant que la transaction 1 reste ouverte, les données de la transaction 2 validée sont invisibles pour les consommateurs.

  Cela a les implications suivantes :

  • Pendant le fonctionnement normal, la latence de visibilité des données est approximativement égale à l'intervalle de checkpoint.

  • Si un job échoue, tout topic sur lequel il écrivait est bloqué pour les consommateurs jusqu'à ce que le job redémarre ou que la transaction expire. Dans les cas extrêmes, le processus d'expiration de la transaction lui-même peut également affecter les opérations de lecture.

FAQ

• Comment récupérer des données JSON depuis Kafka à l'aide de Realtime Compute for Apache Flink ?

• Pourquoi une fenêtre basée sur l'Event Time ne produit-elle aucune sortie pour une table source ApsaraMQ for Kafka ?

• Comment se connecter à un cluster Kafka sécurisé ?

• Comment résoudre les conflits de noms de champs ?

• Workspaces et Namespaces