Realtime Compute for Apache Flink:Kafka YAML connector

Connectez-vous à votre cluster en utilisant l'une des méthodes suivantes :

• Connexion à un cluster ApsaraMQ for Kafka

  • La version du cluster Kafka est 0,11 ou ultérieure.

  • ApsaraMQ for Kafka Le cluster est créé. Pour plus d'informations, consultez la section Créer des ressources.

  • L'espace de travail Flink et le cluster Kafka résident dans le même VPC, et ApsaraMQ for Kafka a ajouté Flink à sa liste d'autorisation. Pour plus d'informations, consultez la section Configurer une liste d'autorisation.

  Important

  Limites liées à l'écriture dans ApsaraMQ for Kafka :

  • ApsaraMQ for Kafka ne prend pas en charge l'écriture de données au format de compression zstd.

  • ApsaraMQ for Kafka ne prend pas en charge les écritures idempotentes ni les écritures transactionnelles ; par conséquent, la sémantique exactly-once fournie par les tables de destination Kafka n'est pas disponible. À partir de la version VVR 8.0.0, le client Kafka open source utilisé par le connecteur Kafka est mis à niveau vers la version 3.x, dans laquelle la propriété properties.enable.idempotence est définie sur true par défaut, ce qui active explicitement les écritures idempotentes. Par conséquent, lorsque vous écrivez dans ApsaraMQ for Kafka avec VVR 8.0.0 ou une version ultérieure, vous devez ajouter explicitement properties.enable.idempotence=false à la table de destination pour désactiver les écritures idempotentes et éviter les échecs d'écriture. Pour comparer les moteurs de stockage et connaître les limites fonctionnelles d'ApsaraMQ for Kafka, consultez la section Comparaison des moteurs de stockage.

• Connexion à un cluster Apache Kafka autogéré

  • La version du cluster Apache Kafka autogéré est 0,11 ou ultérieure.

  • La connectivité réseau entre Flink et le cluster Apache Kafka autogéré est établie. Pour vous connecter à un cluster autogéré via Internet, consultez la section Connexions réseau.

  • Seuls les éléments de configuration client d'Apache Kafka 2.8 sont pris en charge. Pour plus d'informations, consultez la documentation de configuration des consommateurs et des producteurs d'Apache Kafka.

Éléments de configuration

• Général

  Paramètre	Description	Obligatoire	Type de données	Valeur par défaut	Remarques
  type	Le type de la source ou du puits.	Oui	String	Aucune	La valeur doit être kafka.
  name	Le nom de la source ou du puits.	Non	String	Aucune	Aucune
  properties.bootstrap.servers	L'adresse du courtier Kafka.	Oui	String	Aucune	Le format est host:port,host:port,host:port. Utilisez des virgules (,) pour séparer plusieurs adresses.
  properties.*	Les configurations transmises directement au client Kafka.	Non	String	Aucune	Le suffixe doit correspondre à un élément de configuration de producteur ou de consommateur défini dans la documentation officielle d'Apache Kafka. Flink supprime le préfixe properties. et transmet la configuration restante au client Kafka. Par exemple, vous pouvez utiliser 'properties.allow.auto.create.topics' = 'false' pour désactiver la création automatique de topics. Dans la valeur de properties.sasl.jaas.config, vous pouvez définir le nom d'utilisateur et le mot de passe sur secret://kms.<secret-name>.<key> afin de référencer un identifiant géré par KMS au lieu d'utiliser du texte en clair. Cette fonctionnalité est prise en charge à partir de la version VVR 11.9 du moteur Realtime Compute. Pour la procédure de configuration, consultez Intégration de la gestion des identifiants KMS.
  key.format	Le format utilisé pour lire ou écrire la partie clé des messages Kafka.	Non	String	Aucune	Pour les sources, seul json est pris en charge. Pour les puits, valeurs valides : csv json Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.0.0.
  value.format	Le format utilisé pour lire ou écrire la partie valeur des messages Kafka.	Non	String	debezium-json	Pour les sources, valeurs valides : debezium-json canal-json json Pour les puits, valeurs valides : debezium-json canal-json canal-protobuf Remarque Les formats debezium-json et canal-json sont pris en charge uniquement à partir de la version VVR 8.0.10. Le format json est pris en charge uniquement à partir de la version VVR 11.0.0.

• Source

  Paramètre	Description	Obligatoire	Type de données	Valeur par défaut	Remarques
  topic	Le nom du topic à lire.	Non	String	Aucune	Séparez plusieurs noms de topics par des points-virgules (;), par exemple topic-1 et topic-2. Remarque Vous ne pouvez spécifier qu'un seul paramètre entre topic et topic-pattern.
  topic-pattern	L'expression régulière utilisée pour faire correspondre les noms des topics à lire. Tous les topics correspondant à l'expression régulière sont lus lors de l'exécution de la tâche.	Non	String	Aucune	Exemples : user_event_.* : correspond à tous les topics dont le nom commence par user_event_ prod\.logs\..* : correspond aux topics dont le nom a le préfixe prod.logs. (le caractère . doit être échappé) Remarque Vous ne pouvez spécifier qu'un seul paramètre entre topic et topic-pattern.
  properties.group.id	L'ID du groupe de consommateurs.	Non	String	Aucune	Si l'ID de groupe spécifié est utilisé pour la première fois, vous devez définir properties.auto.offset.reset sur earliest ou latest pour spécifier le décalage de démarrage initial.
  scan.startup.mode	Le décalage de démarrage à partir duquel Kafka lit les données.	Non	String	group-offsets	Valeurs valides : earliest-offset : lit les données à partir du décalage le plus ancien de la partition Kafka. latest-offset : lit les données à partir du décalage le plus récent de la partition Kafka. group-offsets (par défaut) : lit les données à partir du décalage validé du groupe de consommateurs spécifié par properties.group.id. timestamp : lit les données à partir de l'horodatage spécifié par scan.startup.timestamp-millis. specific-offsets : lit les données à partir du décalage spécifié par scan.startup.specific-offsets. Remarque Ce paramètre prend effet uniquement lorsque la tâche démarre sans état. Lorsque la tâche redémarre à partir d'un point de contrôle (Checkpoint) ou récupère à partir d'un état, la lecture reprend à partir de la progression enregistrée dans l'état.
  scan.startup.specific-offsets	Le décalage de démarrage de chaque partition lorsque le mode de démarrage est specific-offsets.	Non	String	Aucune	Exemple : partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	L'horodatage de démarrage lorsque le mode de démarrage est timestamp.	Non	Long	Aucune	Unité : millisecondes.
  scan.topic-partition-discovery.interval	L'intervalle de détection dynamique des topics et partitions Kafka.	Non	Duration	5 minutes	L'intervalle de découverte des partitions par défaut est de 5 minutes. Pour désactiver cette fonctionnalité, définissez explicitement l'intervalle de découverte des partitions sur une valeur non positive. Lorsque la découverte dynamique des partitions est activée, la source Kafka détecte automatiquement les nouvelles partitions et lit les données qui s'y trouvent. En mode topic-pattern, la source lit les données des nouvelles partitions des topics existants ainsi que de toutes les partitions des nouveaux topics correspondant à l'expression régulière.
  scan.check.duplicated.group.id	Indique s'il faut vérifier si le groupe de consommateurs spécifié par properties.group.id est dupliqué.	Non	Boolean	false	Valeurs valides : true : vérifie les conflits de groupes de consommateurs avant le démarrage de la tâche. Si un conflit existe, la tâche signale une erreur. Cela permet d'éviter les conflits avec les groupes de consommateurs existants. false : démarre la tâche directement sans vérifier les conflits de groupes de consommateurs.
  schema.inference.strategy	La politique d'analyse du schéma.	Non	String	continuous	Valeurs valides : continuous : analyse le schéma de chaque enregistrement. Lorsque deux schémas consécutifs sont incompatibles, un schéma plus large est analysé et un événement de modification de schéma est généré. static : analyse le schéma une seule fois au démarrage de la tâche. Les données sont ensuite analysées sur la base du schéma initial et aucun événement de modification de schéma n'est généré. Remarque Pour plus d'informations sur l'analyse des schémas, consultez Politiques d'analyse des schémas et de synchronisation des modifications. Ce paramètre est pris en charge uniquement à partir de la version VVR 8.0.11.
  scan.max.pre.fetch.records	Le nombre maximal de messages à consommer et à analyser à partir de chaque partition lors de l'analyse initiale du schéma.	Non	Int	50	Avant que la tâche ne lise et ne traite les données, le nombre spécifié de derniers messages est consommé à l'avance à partir de chaque partition pour initialiser les informations de schéma.
  key.fields-prefix	Le préfixe personnalisé ajouté aux noms de champs analysés à partir de la clé du message. Cela permet d'éviter les conflits de nommage après l'analyse de la clé du message Kafka.	Non	String	Aucune	Par exemple, si ce paramètre est défini sur key_ et que la clé contient un champ nommé a, le champ sera nommé key_a après l'analyse de la clé. Remarque La valeur de key.fields-prefix ne peut pas être un préfixe de value.fields-prefix.
  value.fields-prefix	Le préfixe personnalisé ajouté aux noms de champs analysés à partir de la valeur du message. Cela permet d'éviter les conflits de nommage après l'analyse de la valeur du message Kafka.	Non	String	Aucune	Par exemple, si ce paramètre est défini sur value_ et que la valeur contient un champ nommé b, le champ sera nommé value_b après l'analyse de la valeur. Remarque La valeur de value.fields-prefix ne peut pas être un préfixe de key.fields-prefix.
  metadata.list	Les colonnes de métadonnées à transmettre en aval.	Non	String	Aucune	Les colonnes de métadonnées disponibles sont topic, partition, offset, timestamp, timestamp-type, headers, leader-epoch, __raw_key__ et __raw_value__. Séparez plusieurs colonnes par des virgules (,). Remarque Les colonnes de métadonnées __raw_key__ et __raw_value__ sont disponibles uniquement à partir de la version VVR 11.6.
  scan.value.initial-schemas.ddls	Le schéma initial de tables spécifiques, spécifié à l'aide d'instructions DDL.	Non	String	Aucune	Séparez plusieurs instructions DDL par ;. Par exemple, utilisez CREATE TABLE db1.t1 (id BIGINT, name VARCHAR(10)); CREATE TABLE db1.t2 (id BIGINT); pour spécifier les schémas initiaux des tables db1.t1 et db1.t2. Le schéma de table dans l'instruction DDL doit être cohérent avec la table de destination et respecter les règles de syntaxe Flink SQL. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.5.
  ingestion.ignore-errors	Indique s'il faut ignorer les erreurs lors de l'analyse des données.	Non	Boolean	false	Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.5.
  ingestion.error-tolerance.max-count	Le nombre cumulé d'erreurs d'analyse après lequel la tâche échoue lorsque les erreurs d'analyse des données sont ignorées.	Non	Integer	-1	Ce paramètre prend effet uniquement lorsque ingestion.ignore-errors est activé. La valeur par défaut -1 indique que les exceptions d'analyse ne provoquent pas l'échec de la tâche. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.5.
  scan.duplicate-field.strategy	Spécifie comment gérer les noms de champs en double analysés à partir des parties clé et valeur.	Non	String	EXCEPTION	Valeurs valides : EXCEPTION : lève une exception lorsque des champs en double existent dans la clé et la valeur. Il s'agit du comportement par défaut dans les versions VVR 11.6 et antérieures. PREFER_KEY : privilégie la valeur du champ clé en cas de duplication des champs. PREFER_VALUE : privilégie la valeur du champ valeur en cas de duplication des champs. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.7.

  • Table source au format Debezium JSON

    Paramètre	Obligatoire	Type de données	Valeur par défaut	Description
    debezium-json.distributed-tables	Non	Boolean	false	Activez cette option si les données d'une seule table au format Debezium JSON apparaissent dans plusieurs partitions. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 8.0.11. Important Après avoir modifié ce paramètre, vous devez démarrer la tâche sans état.
    debezium-json.schema-include	Non	Boolean	false	Lorsque vous configurez Debezium Kafka Connect, vous pouvez activer la valeur de configuration Kafka value.converter.schemas.enable pour inclure le schéma dans les messages. Cette option indique si les messages Debezium JSON contiennent le schéma. Valeurs valides : true : les messages Debezium JSON contiennent le schéma. false : les messages Debezium JSON ne contiennent pas le schéma.
    debezium-json.ignore-parse-errors	Non	Boolean	false	Valeurs valides : true : ignore la ligne actuelle lorsqu'une exception d'analyse se produit. false (par défaut) : signale une erreur et la tâche ne parvient pas à démarrer.
    debezium-json.infer-schema.primitive-as-string	Non	Boolean	false	Indique s'il faut analyser tous les types comme String lors de l'analyse du schéma de la table. Valeurs valides : true : analyse tous les types primitifs comme String. false (par défaut) : analyse les données selon les règles de base.
    debezium-json.infer-schema.string-type-inference	Non	Boolean	true	Indique s'il faut tenter de déduire les champs de chaîne comme étant de type TIME, DATE ou TIMESTAMP. Si ce paramètre est défini sur false, l'inférence est ignorée et les champs restent de type STRING. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.8.

  • Table source au format Canal JSON

    Paramètre	Obligatoire	Type de données	Valeur par défaut	Description
    canal-json.distributed-tables	Non	Booléen	false	Activez cette option si les données d'une seule table dans le format Canal JSON apparaissent dans plusieurs partitions. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 8.0.11. Important Après avoir modifié ce paramètre, vous devez démarrer la tâche sans état.
    canal-json.database.include	Non	Chaîne	Aucune	Expression régulière facultative correspondant au champ de métadonnées de la base de données dans les enregistrements Canal, afin de lire uniquement les journaux des modifications (changelog) de la base de données spécifiée. La chaîne d'expression régulière est compatible avec l'objet Java Pattern.
    canal-json.table.include	Non	Chaîne	Aucune	Expression régulière facultative correspondant au champ de métadonnées de la table dans les enregistrements Canal, afin de lire uniquement les journaux des modifications (changelog) de la table spécifiée. La chaîne d'expression régulière est compatible avec l'objet Java Pattern.
    canal-json.ignore-parse-errors	Non	Booléen	false	Valeurs possibles : true : ignore la ligne actuelle en cas d'exception d'analyse. false (par défaut) : signale une erreur et empêche le démarrage de la tâche.
    canal-json.infer-schema.primitive-as-string	Non	Booléen	false	Indique s'il faut analyser tous les types sous forme de String lors de l'analyse du schéma de la table. Valeurs possibles : true : analyse tous les types primitifs sous forme de String. false (par défaut) : analyse les données selon les règles de base.
    canal-json.infer-schema.strategy	Non	Chaîne	AUTO	Politique d'analyse utilisée lors de l'analyse du schéma de la table. Valeurs possibles : AUTO (par défaut) : analyse automatiquement les types en analysant les données JSON. Si les données ne contiennent pas le champ sqlType, nous vous recommandons d'utiliser AUTO pour éviter les échecs d'analyse. SQL_TYPE : analyse les types en se basant sur le tableau sqlType dans les données Canal JSON. Si les données contiennent le champ sqlType, nous vous recommandons de définir canal-json.infer-schema.strategy sur SQL_TYPE pour obtenir des types plus précis. MYSQL_TYPE : analyse les types en se basant sur le tableau mysqlType dans les données Canal JSON. Lorsque les données Canal JSON dans Kafka contiennent le champ sqlType et que vous avez besoin d'un mappage de types plus précis, nous vous recommandons de définir canal-json.infer-schema.strategy sur SQL_TYPE. Pour connaître les règles de mappage des types sqlType, consultez la section Analyse du schéma Canal JSON. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.1. MYSQL_TYPE est pris en charge uniquement à partir de la version VVR 11.3.
    canal-json.mysql.treat-mysql-timestamp-as-datetime-enabled	Non	Booléen	true	Indique s'il faut mapper le type timestamp MySQL au type timestamp CDC : true (par défaut) : mappe le type timestamp MySQL au type timestamp CDC. false : mappe le type timestamp MySQL au type timestamp_ltz CDC.
    canal-json.mysql.treat-tinyint1-as-boolean.enabled	Non	Booléen	true	Indique s'il faut mapper le type MySQL tinyint(1) au type boolean CDC lors de l'utilisation de l'analyse MYSQL_TYPE : true (par défaut) : mappe le type MySQL tinyint(1) au type boolean CDC. false : mappe le type MySQL tinyint(1) au type tinyint(1) CDC. Ce paramètre prend effet uniquement lorsque canal-json.infer-schema.strategy est défini sur MYSQL_TYPE.
    canal-json.infer-schema.string-type-inference	Non	Booléen	true	Indique s'il faut tenter de déduire les champs de type string comme étant de type TIME, DATE ou TIMESTAMP. Si ce paramètre est défini sur false, l'inférence est ignorée et les champs restent de type STRING. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.8.

  • Table source au format JSON

    Paramètre	Obligatoire	Type de données	Valeur par défaut	Description
    json.timestamp-format.standard	Non	Chaîne	SQL	Format d'horodatage en entrée et en sortie. Valeurs possibles : SQL : analyse les horodatages d'entrée au format yyyy-MM-dd HH:mm:ss.s{précision}, par exemple 2020-12-30 12:13:14.123. ISO-8601 : analyse les horodatages d'entrée au format yyyy-MM-ddTHH:mm:ss.s{précision}, par exemple 2020-12-30T12:13:14.123.
    json.ignore-parse-errors	Non	Booléen	false	Valeurs possibles : true : ignore la ligne actuelle en cas d'exception d'analyse. false (par défaut) : signale une erreur et empêche le démarrage de la tâche.
    json.infer-schema.primitive-as-string	Non	Booléen	false	Indique s'il faut analyser tous les types sous forme de String lors de l'analyse du schéma de la table. Valeurs possibles : true : analyse tous les types primitifs sous forme de String. false (par défaut) : analyse les données selon les règles de base.
    json.infer-schema.flatten-nested-columns.enable	Non	Booléen	false	Indique s'il faut développer récursivement les colonnes imbriquées dans les données JSON lors de l'analyse. Valeurs possibles : true : développe récursivement les colonnes imbriquées. false (par défaut) : traite les colonnes imbriquées comme des String.
    json.decode.parser-table-id.fields	Non	Chaîne	Aucune	Indique s'il faut utiliser les valeurs de champs JSON spécifiques pour générer le tableId lors de l'analyse des données JSON. Séparez plusieurs champs par ,. Par exemple, si les données JSON sont {"col0":"a", "col1","b", "col2","c"}, le résultat généré est le suivant : Configuration tableId col0 a col0,col1 a.b col0,col1,col2 a.b.c
    json.infer-schema.fixed-types	Non	Chaîne	Aucune	Types spécifiques de certains champs lors de l'analyse des données JSON. Séparez plusieurs champs par ,. Par exemple, id BIGINT, name VARCHAR(10) spécifie le type du champ id dans les données JSON comme BIGINT et le type du champ name comme VARCHAR(10). Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.5. Lorsque vous utilisez ce paramètre dans VVR 11.5, vous devez également ajouter scan.max.pre.fetch.records: 0.
    json.decode.converter-class	Non	Chaîne	Aucune	Nom complet qualifié de la classe d'implémentation. Le convertisseur peut modifier le tableau d'octets JSON byte[] avant l'analyse des données JSON. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.6.
    json.decode.empty-value-as-delete.enabled	Non	Booléen	false	Indique s'il faut analyser les messages tombstone (avec une valeur vide) dans les topics compactés de Kafka comme des événements DELETE. Cela s'applique aux scénarios où une valeur vide indique une suppression, tels que la mise en miroir de topics compactés et les signaux de suppression CDC. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.7.
    json.infer-schema.string-type-inference	Non	Booléen	true	Indique s'il faut tenter de déduire les champs de type string comme étant de type TIME, DATE ou TIMESTAMP. Si ce paramètre est défini sur false, l'inférence est ignorée et les champs restent de type STRING. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.8.

• Destination (Sink)

  Paramètre	Description	Obligatoire	Type de données	Valeur par défaut	Notes
  type	Type de la destination.	Oui	Chaîne	Aucune	La valeur doit être kafka.
  name	Nom de la destination.	Non	Chaîne	Aucune	Aucune
  topic	Nom du topic Kafka.	Non	Chaîne	Aucune	Si cette option est activée, toutes les données sont écrites dans ce topic. Remarque Si cette option est désactivée, chaque enregistrement est écrit dans le topic nommé d'après sa chaîne TableID (générée en joignant avec .), par exemple databaseName.tableName.
  partition.strategy	Politique utilisée pour écrire les données dans les partitions Kafka.	Non	Chaîne	all-to-zero	Valeurs possibles : all-to-zero (par défaut) : écrit toutes les données dans la partition 0. hash-by-key : écrit les données dans plusieurs partitions en fonction de la valeur de hachage de la clé primaire. Cela garantit que les données ayant la même clé primaire sont écrites dans la même partition dans l'ordre.
  sink.tableId-to-topic.mapping	Mappage entre les noms des tables source et les noms des topics Kafka de destination.	Non	Chaîne	Aucune	Séparez chaque mappage par ;. Séparez le nom de la table source et le nom du topic Kafka de destination par :. Les noms de table peuvent être des expressions régulières. Plusieurs tables mappées vers le même topic peuvent être jointes par ,. Par exemple : mydb.mytable1:topic1;mydb.mytable2:topic2. Remarque Ce paramètre vous permet de modifier le topic mappé tout en conservant les informations du nom de table d'origine.

  • Table de destination au format Canal JSON

    Paramètre	Obligatoire	Type de données	Valeur par défaut	Description
    canal-json.serialize.update.keep-changed-fields-only	Non	Booléen	false	Indique si la partie « ancienne » d'un message UPDATE au format canal-json contient uniquement les valeurs précédentes des champs qui ont changé. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.8.

  • Table de destination au format Debezium JSON

    Paramètre	Obligatoire	Type de données	Valeur par défaut	Description
    debezium-json.include-schema.enabled	Non	Booléen	false	Indique si les données Debezium JSON contiennent des informations de schéma.
    debezium-json.emit.full-table-id.enabled	Non	Booléen	false	Indique s'il faut écrire l'identifiant complet de table en trois parties dans les champs de métadonnées Debezium JSON. Mappage lorsque ce paramètre est activé : Partie de l'identifiant de table CDC Clé Debezium JSON Namespace db Schema schema Table table Mappage lorsque ce paramètre est désactivé : Partie de l'identifiant de table CDC Clé Debezium JSON Namespace Aucune Schema db Table table Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.6.

source:
  type: kafka
  using.built-in-catalog: kafka_catalog

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可选）动态识别每条数据Schema，并比对生成Schema变更
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

source:
  type: kafka
  topic-pattern: user_event_.*
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可选）动态识别每条数据Schema，并比对生成Schema变更
  schema.inference.strategy: continuous

sink:
  type: starrocks
  jdbc-url: jdbc:mysql://<yourFeHostname>:9030
  load-url: <yourFeHostname>:8030
  username: <yourUsername>
  password: ${secret_values.starrocks_password}
 
  # （可选）数据量不大的作业建议调低 flush 间隔，避免数据长时间不落盘（默认 300000，即 5 分钟）
  sink.buffer-flush.interval-ms: 5000
  # （可选）上游为 utf8mb4 字符集时建议设为 4，避免文本截断（默认 3）
  unicode-char.max-bytes: 4
  # （可选）自动建表的分桶数；StarRocks 2.5.7 以下必须显式配置，更高版本可自动推断
  table.create.num-buckets: 8
  # （可选）自动建表的副本数，按集群情况配置
  table.create.properties.replication_num: 3
  # （可选）StarRocks 3.2 及以上建议开启，加速表结构变更
  table.create.properties.fast_schema_evolution: true
  # 注意：通过 transform 变更主键时，必须同时设置 sink.ignore.update-before: false，
  # 否则旧主键对应的行会残留在下游

source:
  type: kafka
  topic: ${kafka.topic}
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  key.format: json
  value.format: json
  # key部分的字段名添加key_前缀
  key.fields-prefix: key_
  # value部分的字段名添加value_前缀
  value.fields-prefix: value_
  # （可选）动态识别每条数据Schema，并比对生成Schema变更
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

source:
  type: kafka
  topic: ${kafka.topic}
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  key.format: json
  value.format: json
  # 优先使用key的字段，忽略同名的value中字段
  scan.duplicate-field.strategy: PREFER_KEY
  # （可选）动态识别每条数据Schema，并比对生成Schema变更
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可选）动态识别每条数据Schema，并比对生成Schema变更
  schema.inference.strategy: continuous
  metadata.list: topic,partition

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可选）动态识别每条数据Schema，并比对生成Schema变更
  schema.inference.strategy: continuous
  # 开启忽略解析报错，默认忽略全部解析报错
  ingestion.ignore-errors: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true
  
# 开启脏数据收集器，打印解析失败数据
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可选）动态识别每条数据Schema，并比对生成Schema变更
  schema.inference.strategy: continuous
  # 开启忽略解析报错
  ingestion.ignore-errors: true
  # 解析报错发生30次后触发作业失败
  ingestion.error-tolerance.max-count: 30
  
sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true
  
# 开启脏数据收集器，打印解析失败数据
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # （可选）动态识别每条数据Schema，并比对生成Schema变更
  schema.inference.strategy: continuous
  # 指定字段 db 和 table 作为 table id
  value.json.decode.parser-table-id.fields: db,tbl

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true
  
# 开启脏数据收集器，打印解析失败数据
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: json
  # 指定字段 db 和 tbl 作为 table id
  value.json.decode.parser-table-id.fields: db,tbl
  # 固定指定部分字段类型
  value.json.infer-schema.fixed-types: 'db STRING, tbl STRING, id BIGINT, amount DECIMAL(18, 2)'
  # 允许未声明字段继续动态推断
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true
  
# 开启脏数据收集器，打印解析失败数据
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: canal-json
  # （可选）动态识别每条数据Schema，并比对生成Schema变更
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true
  
# 开启脏数据收集器，打印解析失败数据
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: canal-json
  # （可选）动态识别每条数据Schema，并比对生成Schema变更
  schema.inference.strategy: continuous
  # 使用 mysql type 信息推导 Schema，也可以配置为 SQL_TYPE 通过 sql type 推导
  value.canal-json.infer-schema.strategy: MYSQL_TYPE

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true
  
# 开启脏数据收集器，打印解析失败数据
pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

source:
  type: kafka
  topic: customers
  properties.bootstrap.servers: localhost:9092
  properties.group.id: ${kafka.group.id}
  value.format: debezium-json
  # （可选）动态识别每条数据Schema，并比对生成Schema变更
  schema.inference.strategy: continuous

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可选）提交用户名，建议为不同作业设置不同的提交用户以避免冲突
  commit.user: your_job_name
  #（可选）开启删除向量，提升读取性能
  table.properties.deletion-vectors.enabled: true

source:
  type: mysql
  hostname: #{hostname}
  port: 3306
  username: #{username}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  #（可选）同步增量阶段新创建的表的数据
  scan.binlog.newly-added-table.enabled: true
  #（可选）同步表注释和字段注释
  include-comments.enabled: true
  #（可选）优先分发无界的分片以避免可能出现的TaskManager OutOfMemory问题
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #（可选）开启解析过滤，加速读取
  scan.only.deserialize.captured.tables.changelog.enabled: true
  # 在 Canal JSON 中补充 mysqlType、sqlType、sql、isDdl 等信息
  include-binlog-meta.enable: true
  
sink:
  type: kafka
  properties.bootstrap.servers: localhost:9092
  topic: order_dw_tables
  # Kafka value 使用 Canal JSON changelog 格式
  value.format: canal-json
  # 指定序列化日期类型数据时使用的格式
  value.canal-json.timestamp-format.standard: SQL
  # 数据统一写入分区0，保证binlog顺序
  partition.strategy: all-to-zero

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 使用数据中的 db、table 字段作为 Table ID
  json.decode.parser-table-id.fields: db,table
  # 设置初始表结构
  scan.value.initial-schemas.ddls: CREATE TABLE db1.t1 (id BIGINT, name VARCHAR(10)); CREATE TABLE db1.t2 (id BIGINT);

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 设置特定字段始终为固定类型
  json.infer-schema.fixed-types: id BIGINT, name VARCHAR(10)
  scan.max.pre.fetch.records: 0

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 开启脏数据容忍功能
  ingestion.ignore-errors: true
  # 容忍 1000 条脏数据
  ingestion.error-tolerance.max-count: 1000

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 开启脏数据容忍功能
  ingestion.ignore-errors: true
  # 容忍所有的脏数据
  ingestion.error-tolerance.max-count: -1

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # 开启脏数据容忍功能
  ingestion.ignore-errors: true
  # 容忍所有的脏数据
  ingestion.error-tolerance.max-count: -1

pipeline:
  dirty-data.collector:
    # 将脏数据写入 TaskManager 的日志文件中
    type: logger

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  
 route:
  - source-table: mydb.mytable1,mydb.mytable2
    sink-table: mytable1

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  sink.tableId-to-topic.mapping: mydb.mytable1,mydb.mytable2:mytable

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  sink.tableId-to-topic.mapping: mydb.mytable1:mytable;mydb.mytable2:mytable

source:                                                                                                                                                                                                                                                                                                            
  type: kafka                                                                                                                                                                                                                                                                                          
  properties.bootstrap.servers: localhost:9092                                                                                                                                                                                                                                                                     
  topic: my_cdc_topic                                                                                                                                                                                                                                                                                              
  properties.group.id: flink-cdc-group                                                                                                                                                                                                                                                                             
  scan.startup.mode: earliest-offset 
  key.format: json  
  value.format: json                                                                                                                                                                                                                                                                            
  # 对 Kafka 消息 key 的 JSON 数据应用自定义转换器
  key.json.decode.converter-class: com.example.KeyExampleConverter                                                                                                                                                                                                                                                                            
  # 对 Kafka 消息 value 的 JSON 数据应用自定义转换器
  value.json.decode.converter-class: com.example.ValueExampleConverter