实时计算Flink版：Kafka数据摄入连接器

配置项

• 通用

  参数	说明	是否必填	数据类型	默认值	备注
  type	源端或目标端类型。	是	String	无	固定值为kafka
  name	源端或目标端名称。	否	String	无	无
  properties.bootstrap.servers	Kafka broker地址。	是	String	无	格式为host:port,host:port,host:port，以英文逗号（,）分割。
  properties.*	对Kafka客户端的直接配置。	否	String	无	后缀名必须是Kafka官方文档中定义的生产者和消费者配置。 Flink会将properties.前缀移除，并将剩余的配置传递给Kafka客户端。例如可以通过'properties.allow.auto.create.topics' = 'false' 来禁用自动创建topic。 其中properties.sasl.jaas.config的用户名和密码可填写secret://kms.<secret-name>.<key>引用KMS托管的凭据，避免明文暴露。仅实时计算引擎VVR 11.9及以上版本支持，配置步骤请参见KMS 凭据管理集成。
  key.format	读取或写入Kafka消息key部分使用的格式。	否	String	无	对Source，仅支持json。 对Sink，取值如下： csv json 说明 仅实时计算引擎VVR 11.0.0及以上版本支持该参数。
  value.format	读取或写入Kafka消息value部分时使用的格式。	否	String	debezium-json	对Source，取值如下： debezium-json  canal-json json 对Sink，取值如下： debezium-json  canal-json canal-protobuf 说明 仅实时计算引擎VVR 8.0.10及以上版本支持debezium-json和canal-json格式。 仅实时计算引擎VVR 11.0.0及以上版本支持json格式。

• 源表

  参数	说明	是否必填	数据类型	默认值	备注
  topic	读取的topic名称。	否	String	无	以英文分号 (;) 分隔多个topic名称，例如topic-1和topic-2 说明 topic和topic-pattern两个选项只能指定其中一个。
  topic-pattern	匹配读取topic名称的正则表达式。所有匹配该正则表达式的topic在作业运行时均会被读取。	否	String	无	示例： user_event_.*：匹配所有以 user_event_ 开头的 topic prod\.logs\..*：匹配 prod.logs. 前缀的 topic（. 需转义） 说明 topic和topic-pattern两个选项只能指定其中一个。
  properties.group.id	消费组ID。	否	String	无	如果指定的group id为首次使用，则必须将properties.auto.offset.reset设置为earliest或latest以指定首次启动位点。
  scan.startup.mode	Kafka读取数据的启动位点。	否	String	group-offsets	取值如下： earliest-offset：从Kafka最早分区开始读取。 latest-offset：从Kafka最新位点开始读取。 group-offsets（默认值）：从指定的properties.group.id已提交的位点开始读取。 timestamp：从scan.startup.timestamp-millis指定的时间戳开始读取。 specific-offsets：从scan.startup.specific-offsets指定的偏移量开始读取。 说明 该参数在作业无状态启动时生效。作业在从checkpoint重启或状态恢复时，会优先使用状态中保存的进度恢复读取。
  scan.startup.specific-offsets	specific-offsets启动模式下，指定每个分区的启动偏移量。	否	String	无	例如partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	timestamp启动模式下，指定启动位点时间戳。	否	Long	无	单位为毫秒
  scan.topic-partition-discovery.interval	动态检测Kafka topic和partition的时间间隔。	否	Duration	5分钟	分区检查间隔默认为5分钟。需要显式地设置分区检查间隔为非正数才能关闭此功能。开启动态分区发现后，Kafka Source 可以自动地发现新增的分区并自动读取对应分区上的数据。在topic-pattern模式下，不仅读取已有topic的新增分区数据，也会读取符合正则匹配的新增topic的所有分区数据。
  scan.check.duplicated.group.id	是否检查通过properties.group.id指定的消费者组有重复。	否	Boolean	false	参数取值如下： true：在启动作业前检查消费者组是否有重复，如有重复作业将会报错，避免与现有的消费者组产生冲突。 false：直接启动作业，不检查消费者组冲突。
  schema.inference.strategy	Schema解析策略。	否	String	continuous	取值如下： continuous：对每条数据均进行 Schema 解析。在前后 Schema 不兼容时，解析出更宽的 Schema 并产生 Schema 变更事件。 static：仅在作业启动时进行一次Schema解析，后续根据初始Schema解析数据，不会产生Schema变更事件。 说明 Schema解析详情可见Kafka SQL连接器。 仅VVR 8.0.11及以上版本支持该配置项。
  scan.max.pre.fetch.records	Schema初始解析时，对每个分区最多尝试消费解析的消息数量	否	Int	50	在作业实际读取并处理数据前，对每个分区尝试提前消费指定数量的最新消息，用于初始化Schema信息。
  key.fields-prefix	自定义添加到消息键（Key）解析出字段名称的前缀，以避免Kafka消息键解析后的命名冲突问题。	否	String	无	假设该配置项设为key_，当key中包含字段名a时，解析key后该字段名称为key_a。 说明 key.fields-prefix的配置值不可以是value.fields-prefix的前缀。
  value.fields-prefix	自定义添加到消息体（Value）解析出字段名称的前缀，以避免Kafka消息体解析后的命名冲突问题。	否	String	无	假设该配置项设为value_，当value中包含字段名b时，解析value后该字段名称为value_b。 说明 value.fields-prefix的配置值不可以是key.fields-prefix的前缀。
  metadata.list	需要传递给下游的元数据列	否	String	无	可用的元数据列包括topic、partition、offset、timestamp、timestamp-type、headers、leader-epoch、__raw_key__和__raw_value__。使用英文逗号分隔。 说明 __raw_key__和 __raw_value__元数据列需要VVR 11.6及更高版本可用。
  scan.value.initial-schemas.ddls	通过DDL方式指定某些表的初始Schema。	否	String	无	多个DDL使用英文;连接。例如，使用CREATE TABLE db1.t1 (id BIGINT, name VARCHAR(10)); CREATE TABLE db1.t2 (id BIGINT);为表db1.t1和表db1.t2分别指定初始Schema。 这里的DDL表结构应当和写入目标表保持一致，并且满足Flink SQL的语法规则。 说明 VVR 11.5及以上版本支持该配置。
  ingestion.ignore-errors	是否忽略数据解析过程中的报错。	否	Boolean	false	说明 VVR 11.5及以上版本支持该配置。
  ingestion.error-tolerance.max-count	忽略数据解析过程报错时，累计多少次报错后作业失败。	否	Integer	-1	仅当ingestion.ignore-errors开启时生效，默认值-1表示解析异常不触发作业失败。 说明 VVR 11.5及以上版本支持该配置。
  scan.duplicate-field.strategy	在 Key 和 Value 部分解析出重复字段名时，应该如何进行处理。	否	String	EXCEPTION	参数取值如下： EXCEPTION：在 key 和 value 存在重复字段时，抛出异常。这是 VVR 11.6 及更早版本的默认行为。 PREFER_KEY：在字段重复时，优先选择 key 字段的值。 PREFER_VALUE：在字段重复时，优先选择 value 字段的值。 说明 VVR 11.7 及以上版本支持该配置。
  table-id.include	按解析后 Table ID 选择需要保留的表，默认保留全部表。	否	String	无	支持正则表达式，点号需要转义，如inventory.orders_\.*表示保留 inventory 库内以 orders_ 开头的表。 多个选择器用英文逗号分隔，匹配任意一个即保留。 说明 VVR 11.9 及以上版本支持该配置。
  scan.emit.create-table-events.in-batch.enabled	是否批量发送初始建表事件。	否	Boolean	true	仅在如下2种场景生效，其他情况会自动关闭该能力。 未配置key format 配置了key format，value format为json且未配置decode.parser-table-id.fields 说明 VVR 11.9 及以上版本支持该配置。

  • 源表 Debezium JSON 格式

    参数	是否必填	数据类型	默认值	描述
    debezium-json.distributed-tables	否	Boolean	false	如果Debezium JSON内单张表数据会出现在多个分区，则需要开启此选项。 说明 仅VVR 8.0.11及以上版本支持该配置项。 重要 修改该配置项后，需要无状态启动作业。
    debezium-json.schema-include	否	Boolean	false	设置Debezium Kafka Connect时，可以启用Kafka配置value.converter.schemas.enable，以在消息中包含schema。此选项表明Debezium JSON消息是否包含schema。 参数取值如下： true：Debezium JSON消息包含schema。 false：Debezium JSON消息不包含schema。
    debezium-json.ignore-parse-errors	否	Boolean	false	参数取值如下： true：当解析异常时，跳过当前行。 false（默认值）：报出错误，作业启动失败。
    debezium-json.infer-schema.primitive-as-string	否	Boolean	false	解析表结构时，是否解析所有类型为String类型。 参数取值如下： true：解析所有基本类型为String。 false（默认值）：按照基本规则进行解析。
    debezium-json.infer-schema.string-type-inference	否	Boolean	true	是否尝试将字符串字段推断为 TIME、DATE 或 TIMESTAMP类型。设置为 false 时跳过该推断，字段保持 STRING。 说明 VVR 11.8及以上版本支持该配置。

  • 源表 Canal JSON 格式

    参数	是否必填	数据类型	默认值	描述
    canal-json.distributed-tables	否	Boolean	false	如果Canal JSON内单张表数据会出现在多个分区，则需要开启此选项。 说明 仅VVR 8.0.11及以上版本支持该配置项。 重要 修改该配置项后，需要无状态启动作业。
    canal-json.database.include	否	String	无	一个可选的正则表达式，通过正则匹配Canal记录中的database元字段，仅读取指定数据库的changelog记录。正则字符串与Java的Pattern兼容。
    canal-json.table.include	否	String	无	一个可选的正则表达式，通过正则匹配Canal记录中的table元字段，仅读取指定表的changelog记录。正则字符串与Java的Pattern兼容。
    canal-json.ignore-parse-errors	否	Boolean	false	参数取值如下： true：当解析异常时，跳过当前行。 false（默认值）：报出错误，作业启动失败。
    canal-json.infer-schema.primitive-as-string	否	Boolean	false	解析表结构时，是否所有类型解析为String类型。 参数取值如下： true：解析所有基本类型为String。 false（默认值）：按照基本规则进行解析。
    canal-json.infer-schema.strategy	否	String	AUTO	解析表结构时的解析策略。 参数取值如下： AUTO（默认值）：通过解析JSON数据自动解析。如果数据中不包含sqlType字段，建议使用AUTO以避免解析失败。 SQL_TYPE：通过canal json数据中的sqlType数组解析。如果数据包含sqlType字段时，建议将canal-json.infer-schema.strategy设置为SQL_TYPE以获得更精确的类型。 MYSQL_TYPE：通过canal json数据中的mysqlType数组解析。 当Kafka中的Canal JSON数据包含sqlType字段且需要更精确的类型映射时，建议将canal-json.infer-schema.strategy设置为SQL_TYPE。 sqlType类型映射规则请见Kafka SQL连接器。 说明 VVR 11.1及以上版本支持该配置。 MYSQL_TYPE在VVR 11.3及以上版本支持使用。
    canal-json.mysql.treat-mysql-timestamp-as-datetime-enabled	否	Boolean	true	是否将mysql timestamp类型映射到cdc timestamp类型： true（默认）：mysql timestamp类型映射到cdc timestamp类型。 false：mysql timestamp类型映射到cdc timestamp_ltz类型。
    canal-json.mysql.treat-tinyint1-as-boolean.enabled	否	Boolean	true	使用MYSQL_TYPE解析时，是否将mysql tinyint(1)类型映射到cdc boolean类型： true（默认）：mysql tinyint(1)类型映射到cdc boolean类型。 false：mysql tinyint(1)类型映射到cdc tinyint(1)类型。 该配置仅当canal-json.infer-schema.strategy配置为MYSQL_TYPE时生效。
    canal-json.infer-schema.string-type-inference	否	Boolean	true	是否尝试将字符串字段推断为 TIME、DATE 或 TIMESTAMP类型。设置为 false 时跳过该推断，字段保持 STRING。 说明 VVR 11.8及以上版本支持该配置。

  • 源表 JSON 格式

    参数	是否必填	数据类型	默认值	描述
    json.timestamp-format.standard	否	String	SQL	指定输入和输出时间戳格式。参数取值如下： SQL：解析yyyy-MM-dd HH:mm:ss.s{precision}格式的输入时间戳，例如2020-12-30 12:13:14.123。 ISO-8601：解析yyyy-MM-ddTHH:mm:ss.s{precision}格式的输入时间戳，例如2020-12-30T12:13:14.123。
    json.ignore-parse-errors	否	Boolean	false	参数取值如下： true：当解析异常时，跳过当前行。 false（默认值）：报出错误，作业启动失败。
    json.infer-schema.primitive-as-string	否	Boolean	false	解析表结构时，是否解析所有类型为String类型。 参数取值如下： true：解析所有基本类型为String。 false（默认值）：按照基本规则进行解析。
    json.infer-schema.flatten-nested-columns.enable	否	Boolean	false	解析JSON格式数据时，是否递归式地展开JSON中的嵌套列。参数取值如下： true：递归式展开。 false（默认值）：将嵌套列当作String处理。
    json.decode.parser-table-id.fields	否	String	无	解析JSON格式数据时，是否使用部分JSON字段值生成tableId，多个字段使用英文,连接。例如：JSON数据为{"col0":"a", "col1","b", "col2","c"}，生成结果如下： 配置 tableId col0 a col0,col1 a.b col0,col1,col2 a.b.c
    json.infer-schema.fixed-types	否	String	无	解析JSON格式数据时，指定某些字段的具体类型，多个字段使用英文,连接。例如：id BIGINT, name VARCHAR(10)可以指定JSON数据中的id字段类型为BIGINT，name字段类型为VARCHAR(10)。 说明 VVR 11.5及以上版本支持该配置。 VVR 11.5版本使用该配置时，需要额外添加配置scan.max.pre.fetch.records: 0。
    json.decode.converter-class	否	String	无	在解析JSON数据前，通过converter可以对JSON byte[]提前进行修改，填写具体实现类的全限定名。 说明 VVR 11.6及以上版本支持该配置。
    json.decode.empty-value-as-delete.enabled	否	Boolean	false	是否将Kafka compacted topic中的墓碑消息（value 为空）解析为 DELETE 事件，用于 compacted topic 镜像、CDC 删除信号等以空 value 表示删除语义的场景。 说明 VVR 11.7 及以上版本支持该配置。
    json.infer-schema.string-type-inference	否	Boolean	true	是否尝试将字符串字段推断为 TIME、DATE 或 TIMESTAMP类型。设置为 false 时跳过该推断，字段保持 STRING。 说明 VVR 11.8及以上版本支持该配置。

• 结果表

  参数	说明	是否必填	数据类型	默认值	备注
  type	目标端类型。	是	String	无	固定值为kafka
  name	目标端名称。	否	String	无	无
  topic	Kafka Topic名称。	否	String	无	开启时，所有的数据都会写入这个Topic。 说明 如果没有开启，每条数据会写入到其TableID对应字符串（通过.拼接生成）的Topic，例如databaseName.tableName。
  partition.strategy	数据写入Kafka分区的策略。	否	String	all-to-zero	取值如下： all-to-zero(默认值):将所有数据写入 0 号分区。 hash-by-key:根据主键的哈希值将数据写到多个分区。保证同一个主键的数据在同一个分区并且有序。
  sink.tableId-to-topic.mapping	上游表名到下游 Kafka Topic名的映射关系。	否	String	无	每个映射关系由;分割，上游表的表名和下游 Kafka 的Topic名由:分割，表名可以使用正则表达式，映射到同一个topic的多张表可以使用,拼接。例如mydb.mytable1:topic1;mydb.mytable2:topic2。 说明 配置这个参数能够在保留原始表名信息的同时修改映射的Topic。

  • 结果表Canal JSON格式

    参数	是否必填	数据类型	默认值	描述
    canal-json.serialize.update.keep-changed-fields-only	否	Boolean	false	写入 canal-json 格式的 UPDATE 消息时，old 部分是否仅包含发生变化字段的变更前的值。 说明 VVR 11.8及以上版本支持该配置。

  • 结果表Debezium JSON格式

    参数	是否必填	数据类型	默认值	描述
    debezium-json.include-schema.enabled	否	Boolean	false	Debezium JSON数据中是否包含Schema信息。
    debezium-json.emit.full-table-id.enabled	否	Boolean	false	是否将完整的三段式Table ID写入Debezium JSON元数据字段中。 启用此参数时的映射关系为： CDC Table ID部分 Debezium JSON键 Namespace db Schema schema Table table 禁用此参数时的映射关系为： CDC Table ID部分 Debezium JSON键 Namespace 无 Schema db Table table 说明 VVR 11.6及以上版本支持此配置。

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

读取单独topic

读取topic customers，写入阿里云数据湖的配置示例：

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

其中，Json格式默认生成的表名和topic名称相同。

读取多个topic

读取符合正则表达式的多个topic，写入 StarRocks SQL连接器 的配置示例：

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，
  # 否则旧主键对应的行会残留在下游

其中，Json格式默认生成的表名和topic名称相同。

读取key并防止字段冲突

为了防止 key 和 value 的字段冲突导致报错，有以下解决方案：

1. 为字段添加前缀，避免字段冲突：

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

2. 配置冲突解决策略，详情请见Kafka SQL连接器。如下配置会优先使用key中字段，出现重复时忽略value中的同名字段。

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

增加元数据列

读取topic customers，并在字段增加元数据topic和partition的配置示例：

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

累计解析失败30次后作业报错的配置示例：

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

JSON数据读取

接下来介绍 Json 格式读取常见的一些使用方式。

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

Canal Json数据读取

接下来介绍 Canal Json 格式读取常见的一些使用方式。

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

Debezium Json数据读取

读取topic customers，写入阿里云数据湖的配置示例：

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

MySQL 原始 binlog 同步 Kafka

Flink CDC数据摄入支持同步MySQL 原始binlog内容到 canal json，如下作业可以同步多张表的binlog到 topic order_dw_tables。

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

表结构解析和变更同步策略

Kafka连接器会维护当前已知的所有表的Schema。

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

脏数据容忍与收集

在某些情况下，您的 Kafka 数据源中可能包含一些格式不正确的数据（脏数据），为了避免同步作业因为这些脏数据导致频繁失败重启，您可以配置忽略这部分异常数据。配置示例如下：

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

这里配置了忽略 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

脏数据容忍策略保障了作业不会因为异常数据频繁失败，您可能还希望进一步的了解这些脏数据的信息以调整 Kafka 数据生产者的行为，参考脏数据收集中介绍的流程，您可以在 TaskManager 日志中查看到作业的脏数据，配置示例如下：

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

表名与Topic的映射策略

在使用kafka作为数据摄入作业的目标端时，由于写入到Kafka消息格式（debezium-json或者canal-json）中还包含表名信息，后续消费Kafka消息时往往以数据中的表名信息作为实际表名（而非topic名称），因此需要谨慎配置表名与Topic的映射策略。

假设在MySQL有mydb.mytable1,mydb.mytable2两张表需要同步，可能的配置策略有以下几种：

1. 不配置任何映射策略

在没有任何映射策略的情况下，每张表会写入到对应的由“库名.表名”组成的topic中。因此mydb.mytable1的数据会写入到名为mydb.mytable1的topic中，mydb.mytable2的数据会写入到名为mydb.mytable2的topic中。配置示例如下：

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}

2. 配置route规则进行映射（不推荐）

在很多场景下，用户不希望写入的topic直接为“库名.表名”的格式，希望将数据写入到指定的topic中，因此会配置route规则进行映射。配置示例如下：

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

此时所有来自mydb.mytable1,mydb.mytable2的数据都会写入到mytable1这一个topic中。

然而，通过route规则修改写入的topic名称时，也会修改Kafka消息（debezium-json或者canal-json格式）中的表名信息，此时Kafka消息中所有的表名都为mytable1，在其他系统消费这个topic的Kafka消息时，可能出现不符合预期的情况。

3. 配置sink.tableId-to-topic.mapping参数进行映射（推荐）

为了在配置表名与Topic的映射规则的同时保留源表表名信息，可以使用sink.tableId-to-topic.mapping参数完成该需求。配置示例如下：

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

此时所有来自mydb.mytable1,mydb.mytable2的数据都会写入到mytable1这一个topic中，并且Kafka消息（debezium-json或者canal-json格式）中的表名信息仍然为mydb.mytable1或者mydb.mytable2，在其他系统消费这个topic的Kafka消息时，能正确获取源表表名信息。

JSON Converter实现

JSON格式的数据有时无法保证格式完全一致，或者有处理需求无法通过Transform配置模块完成，例如合并两个列为新的列并删除原有的两列，并保证能进行 Schema 变更同步。为了更灵活处理JSON数据，可以通过实现KafkaPayloadConverter接口在框架处理前，提前对JSON数据进行修改，使用时需要添加配置json.decode.converter-class填写具体实现类的全限定名。

实现和使用JSON Converter的流程如下：

1. 实现KafkaPayloadConverter接口并打包，在开源Demo仓库已经提供了一些实现示例。

2. 将打好的包添加到数据摄入作业的附加依赖。

3. 在Source模块填写json.decode.converter-class配置，以开源项目中的ArrayElementExtractorConverter类为例：

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

4. 部署作业并运行。

key format和value format

在VVR 11.5及以前的版本中，key format 和 value format的配置无法进行区分，当使用相同的format时，format配置会同时应用在key format和value format。

VVR 11.6及以后的版本对这部分进行了优化，以json format 为例，format 的配置项的传递规则如下：

• format前缀（如json.infer-schema.primitive-as-string）： 为了保持版本兼容，format前缀的配置会默认同时应用到 key format和 value format。

• key前缀加format前缀（如key.json.infer-schema.primitive-as-string）： key前缀加format前缀的配置会只应用在key format，优先级比format前缀的配置更高。

• value前缀加format前缀（如value.json.infer-schema.primitive-as-string）： value前缀加format前缀的配置会只应用在value format，优先级比format前缀的配置更高。

如下的Kafka source配置，为key format和value format配置了不同的JSON Converter。

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