Kafka 连接器既可以作为 Source（
type: kafka）从 Kafka 主题读取变更事件，也可以作为 Sink（type: kafka）把变更事件写入 Kafka 主题。
作为 Source 时，连接器支持解析
debezium-json、canal-json 以及普通 json 三种消息格式，并支持自动推断表结构。作为 Sink 时，支持以
debezium-json、canal-json 格式写出，支持自定义投递分区、消息头与表名到主题的映射。
前置准备
Kafka 集群的 broker 地址（bootstrap servers）对作业运行环境可访问。
作为 Source 时，需要保证账号具备读取目标主题的消费权限；作为 Sink 时，需要具备写入目标主题的生产权限。
作为 Source 时，
topic 与 topic-pattern 必须且只能配置其中一个。多个作业消费同一个主题时，建议显式配置不同的 consumer group id，避免位点互相干扰。
版本说明
Flink 版本 | 说明 |
1.20 | 基于 Flink-cdc 社区 release-3.6 开发，JDK 版本为 11，支持 Apache Kafka 消息队列 |
使用范围
可作为 Source（
type: kafka）与 Sink（type: kafka）使用。
快速示例
MySQL 同步到 Kafka
source:type: mysqlname: MySQL Sourcehostname: 127.0.0.1port: 3306username: adminpassword: passtables: adb.\\.*server-id: 5401-5404sink:type: kafkaname: Kafka Sinkproperties.bootstrap.servers: 127.0.0.1:9092topic: cdc-topicvalue.format: debezium-jsonpipeline:name: MySQL to Kafka Pipelineparallelism: 4
Kafka 同步到 Doris
source:type: kafkaname: Kafka Sourceproperties.bootstrap.servers: 127.0.0.1:9092topic: cdc-topicvalue.format: debezium-jsonscan.startup.mode: group-offsetssink:type: dorisname: Doris Sinkfenodes: 127.0.0.1:8030username: rootpassword: passpipeline:name: Kafka to Doris Pipelineparallelism: 4
Source 配置项
通用配置项
参数 | 是否必填 | 默认值 | 类型 | 说明 |
properties.bootstrap.servers | required | (none) | String | Kafka 集群的 broker 地址，例如 127.0.0.1:9092。 |
topic | 与 topic-pattern 二选一 | (none) | String | 需要消费的主题名，多个主题之间用英文分号（;）分隔。 与 topic-pattern 必须且只能配置其中一个。 |
topic-pattern | 与 topic 二选一 | (none) | String | 需要消费的主题名正则表达式。 与 topic 必须且只能配置其中一个。 |
value.format | optional | debezium-json | String | 消息体的格式，可取 debezium-json、canal-json、json。 未显式配置时会回退读取 format 参数，两者都未配置时为 debezium-json。 取值大小写不敏感。 |
format | optional | (none) | String | 消息体格式的别名参数，仅在未配置 value.format 时生效。 |
key.format | optional | (none) | String | 是否解析消息键。当前仅支持 json，配置其它值会报错。 未配置时不解析消息键。 |
key.fields-prefix | optional | (none) | String | 为消息键中的字段统一添加的前缀，用于避免键与值中的同名字段冲突。 仅在配置了 key.format 时可用。 |
value.fields-prefix | optional | (none) | String | 为消息体中的字段统一添加的前缀。 与 key.fields-prefix 不能互为前缀。 |
scan.startup.mode | optional | group-offsets | String | 启动模式，可取 group-offsets、earliest-offset、 latest-offset、timestamp、specific-offsets， 详见「启动模式」章节。 取值大小写不敏感。 |
scan.startup.timestamp-millis | optional | (none) | Long | timestamp 启动模式下的毫秒时间戳，该模式下必填。 |
scan.startup.specific-offsets | optional | (none) | String | specific-offsets 启动模式下的分区位点，该模式下必填。 格式为 partition:0,offset:42，多个分区之间用英文分号（;）分隔。 |
scan.topic-partition-discovery.interval | optional | 5min | Duration | 检查并发现新分区的间隔。 |
scan.header-filter | optional | (none) | String | 消息头的过滤条件。 |
scan.check.duplicated.group.id | optional | false | Boolean | 是否检查 consumer group id 被多个作业重复使用。 |
scan.max.pre.fetch.records | optional | 50 | Integer | 推断表结构时最多预读的记录数，必须大于等于 0。设为 0 表示不预读。 |
scan.value.initial-schemas.ddls | optional | (none) | String | 用于初始化表结构的 Flink SQL DDL 语句集合。 |
scan.duplicate-field.strategy | optional | exception | String | 键与值中出现同名字段时的处理策略，可取 exception（报错）或 prefer-key（保留键中的字段）。 取值大小写不敏感。 |
schema.inference.strategy | optional | continuous | String | 表结构推断策略，可取 continuous（持续推断）或 static（只推断一次）。 取值大小写不敏感。 |
metadata.list | optional | (none) | String | 需要额外透传到下游的元数据列表，英文逗号分隔， 详见「支持的元数据列」章节。 |
ingestion.ignore-errors | optional | false | Boolean | 是否忽略数据接入过程中的解析错误。开启后解析失败的记录会被跳过。 与当前格式对应的 ignore-parse-errors 满足其一即生效， 无需同时配置。 |
ingestion.error-tolerance.max-count | optional | -1 | Long | 允许出现的最大错误条数。设为 -1 表示不限制。 |
properties.* | optional | (none) | String | 透传给 Kafka 客户端的原生配置，例如 properties.group.id: cdc-group、 properties.auto.offset.reset: earliest。 属性名去掉该前缀后传给 Kafka 客户端。 |
未配置
properties.group.id 时，连接器会根据订阅的主题或主题模式自动生成一个稳定的 group id。
debezium-json 配置项
以下参数仅在
value.format 为 debezium-json 时生效。参数 | 是否必填 | 默认值 | 类型 | 说明 |
debezium-json.distributed-tables | optional | false | Boolean | 是否把 Debezium 消息按分库分表方式展开为多张表。 |
debezium-json.schema-include | optional | false | Boolean | 是否包含 schema 信息。 |
debezium-json.ignore-parse-errors | optional | false | Boolean | 是否忽略解析错误。 |
debezium-json.infer-schema.primitive-as-string | optional | false | Boolean | 是否把推断出的基本类型统一按 STRING 处理。 |
canal-json 配置项
以下参数仅在
value.format 为 canal-json 时生效。参数 | 是否必填 | 默认值 | 类型 | 说明 |
canal-json.distributed-tables | optional | false | Boolean | 是否把 Canal 消息按分库分表方式展开为多张表。 |
canal-json.database.include | optional | (none) | String | 只同步匹配的 database。 |
canal-json.table.include | optional | (none) | String | 只同步匹配的表。 |
canal-json.ignore-parse-errors | optional | false | Boolean | 是否忽略解析错误。 |
canal-json.infer-schema.primitive-as-string | optional | false | Boolean | 是否把推断出的基本类型统一按 STRING 处理。 |
canal-json.infer-schema.strategy | optional | auto | String | 类型推断策略，可取 auto 或 sql-type。 取值大小写不敏感。 |
canal-json.mysql.treat-mysql-timestamp-as-datetime-enabled | optional | true | Boolean | 是否把 MySQL 的 TIMESTAMP 类型按 DATETIME 处理。 |
canal-json.mysql.treat-tinyint1-as-boolean.enabled | optional | true | Boolean | 是否把 MySQL 的 TINYINT(1) 当作 BOOLEAN 处理。 |
json 配置项
以下参数仅在
value.format 为 json 时生效。参数 | 是否必填 | 默认值 | 类型 | 说明 |
json.ignore-parse-errors | optional | false | Boolean | 是否忽略 JSON 解析错误。 |
json.infer-schema.primitive-as-string | optional | false | Boolean | 是否把推断出的基本类型统一按 STRING 处理。 |
json.infer-schema.flatten-nested-columns.enable | optional | false | Boolean | 是否把嵌套结构摊平为多列。 |
json.timestamp-format.standard | optional | sql | String | 指定时间字符串的解析格式标准。 |
json.decode.parser-table-id.fields | optional | (none) | String | 指定用消息中哪些字段拼成表 ID，英文逗号分隔，字段个数必须为 1 到 3。 |
json.infer-schema.fixed-types | optional | (none) | String | 强制指定某些字段的类型，跳过自动推断。格式为 字段名 类型， 多条之间用英文逗号分隔。 |
json.decode.converter-class | optional | (none) | String | 自定义 JSON 消息转换器的全限定类名，该类需实现 KafkaPayloadConverter 并提供 public 的无参构造方法。 |
启动模式
scan.startup.mode 决定作业从 Kafka 的哪个位置开始消费（取值大小写不敏感）：
group-offsets（默认）：从 consumer group 已提交的位点开始消费。earliest-offset：从分区的最早位点开始消费。latest-offset：从分区的最新位点开始消费，即只消费作业启动之后产生的消息。timestamp：从指定的时间戳开始消费，必须同时配置 scan.startup.timestamp-millis。specific-offsets：从指定的分区位点开始消费，必须同时配置 topic（不能用 topic-pattern）与
scan.startup.specific-offsets。
支持的元数据列
通过
metadata.list 可以指定透传到下游的元数据，可在 transform 中直接引用。多个值之间用英文逗号分隔。Key | 类型 | 说明 |
topic | STRING | 消息所在的 Kafka 主题名。 |
partition | INT | 消息所在的分区号。 |
headers | MAP(STRING, BYTES) | 消息的 header 集合。 |
leader-epoch | INT | 消息所在的 leader epoch。 |
offset | BIGINT | 消息在分区中的偏移量。 |
timestamp | TIMESTAMP_LTZ(3) | 消息的时间戳。 |
timestamp-type | STRING | 消息的时间戳类型。 |
__raw_key__ | STRING | 原始消息键。 |
__raw_value__ | STRING | 原始消息体。 |
配置示例：
source:type: kafkametadata.list: topic,partition,offset,timestamp
配置
metadata.list 时只会透传其中列出的列，且列名必须是上表中的值，否则作业启动报错。
Sink 配置项
参数 | 是否必填 | 默认值 | 类型 | 说明 |
properties.bootstrap.servers | required | (none) | String | Kafka 集群的 broker 地址，例如 127.0.0.1:9092。 缺少该参数时 Kafka 客户端无法连接集群，必须配置。 |
topic | optional | (none) | String | 配置后所有事件都会发送到该主题。 未配置时按上游表名自动推断目标主题。 |
sink.tableId-to-topic.mapping | optional | (none) | String | 上游表到目标主题的自定义映射。 每条映射用英文冒号（:）分隔表选择器与主题名，多条映射之间用英文分号（;）分隔。 例如 mydb.mytable1:topic1;mydb.mytable2:topic2。 |
value.format | optional | debezium-json | String | 消息体的格式，可取 debezium-json 或 canal-json。 取值大小写不敏感。 |
key.format | optional | json | String | 消息键的格式，可取 json 或 csv。 取值大小写不敏感。 |
partition.strategy | optional | all-to-zero | String | 分区投递策略。 all-to-zero：所有记录都发往 0 号分区； hash-by-key：按消息键哈希选择分区。 取值大小写不敏感。 |
sink.delivery-guarantee | optional | at-least-once | String | 投递语义。 at-least-once：数据确保送达，但可能重复； exactly-once：精确一次； none：尽力而为，速度最快，但可能丢失数据。 取值大小写不敏感。 |
sink.add-tableId-to-header-enabled | optional | false | Boolean | 是否为每条记录添加 namespace、schemaName、 tableName 三个 header。 |
sink.custom-header | optional | (空) | String | 为每条记录附加的自定义 header。 每个 header 的键值用英文冒号（:）分隔，多个 header 之间用英文分号（;）分隔。 例如 key1:value1;key2:value2。 |
debezium-json.include-schema.enabled | optional | false | Boolean | 仅 value.format=debezium-json 时生效， 是否在每条记录中携带 Debezium 的 schema 信息。 |
properties.* | optional | (none) | String | 透传给 Kafka 客户端的原生配置。 属性名去掉该前缀后传给 Kafka 客户端。 |
Sink 侧的时区取自 pipeline 配置
local-time-zone，未配置时使用作业运行环境的系统时区。
