Kafka Pipeline 连接器

Kafka Pipeline 连接器可以用作 Pipeline 的 Data Sink，将数据写入 Kafka。本文档介绍如何设置 Kafka Pipeline 连接器。

连接器的功能

• 自动建表
• 表结构变更同步
• 数据实时同步

如何创建 Pipeline

从 MySQL 读取数据同步到 Kafka 的 Pipeline 可以定义如下：

source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
  server-id: 5401-5404

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: PLAINTEXT://localhost:62510

pipeline:
  name: MySQL to Kafka Pipeline
  parallelism: 2

Pipeline 连接器配置项

Option	Required	Default	Type	Description
type	required	(none)	String	指定要使用的连接器，这里需要设置成 'kafka'。
name	optional	(none)	String	Sink 的名称。
partition.strategy	optional	(none)	String	定义发送数据到 Kafka 分区的策略，可以设置的选项有 `all-to-zero`（将所有数据发送到 0 号分区）以及 `hash-by-key`（所有数据根据主键的哈希值分发），默认值为 `all-to-zero`。
key.format	optional	(none)	String	用于序列化 Kafka 消息的键部分数据的格式。可以设置的选项有 `csv` 以及 `json`，默认值为 `json`。
value.format	optional	(none)	String	用于序列化 Kafka 消息的值部分数据的格式。可选的填写值包括 debezium-json 和 canal-json, 默认值为 `debezium-json`，并且目前不支持用户自定义输出格式。
properties.bootstrap.servers	required	(none)	String	用于建立与 Kafka 集群初始连接的主机/端口对列表。
topic	optional	(none)	String	如果配置了这个参数，所有的消息都会发送到这一个主题。
sink.add-tableId-to-header-enabled	optional	(none)	Boolean	如果配置了这个参数，所有的消息都会带上键为 `namespace`, 'schemaName', 'tableName'，值为事件 TableId 里对应的 字符串的 header。
properties.*	optional	(none)	String	将 Kafka 支持的参数传递给 pipeline，参考 Kafka consume options。
sink.custom-header	optional	(none)	String	Kafka 记录自定义的 Header。每个 Header 使用 ',' 分割，键值使用 ':' 分割。举例来说，可以使用这种方式 'key1:value1,key2:value2'。
sink.tableId-to-topic.mapping	optional	(none)	String	自定义的上游表名到下游 Kafka Topic 名的映射关系。每个映射关系由 `;` 分割，上游表的 TableId 和下游 Kafka 的 Topic 名由 `:` 分割。举个例子，我们可以配置 `sink.tableId-to-topic.mapping` 的值为 `mydb.mytable1:topic1;mydb.mytable2:topic2`。
sink.delivery-guarantee	optional	at-least-once	String	提交时的投递语义。可选值为 `exactly-once`、`at-least-once` 和 `none`，默认值为 `at-least-once`。
debezium-json.include-schema.enabled	optional	false	Boolean	如果配置了这个参数，每条 debezium 记录都将包含 debezium schema 信息。只有当 `value.format` 为 `debezium-json` 时才生效。

使用说明

• 写入 Kafka 的 topic 默认会是上游表 namespace.schemaName.tableName 对应的字符串，可以通过 pipeline 的 route 功能进行修改。
• 如果配置了 topic 参数，所有的消息都会发送到这一个主题。
• 写入 Kafka 的 topic 如果不存在，则会默认创建。

输出格式

对于不同的内置 value.format 选项，输出的格式也是不同的：

debezium-json

参考 Debezium docs，debezium-json 格式会包含 before,after,op,source 几个元素，但是 ts_ms 字段并不会包含在 source 元素中。
一个输出的示例是：

{
  "before": null,
  "after": {
    "col1": "1",
    "col2": "1"
  },
  "op": "c",
  "source": {
    "db": "default_namespace",
    "table": "table1"
  }
}

当 debezium-json.include-schema.enabled=true 时，输出示例如下：

{
  "schema":{
    "type":"struct",
    "fields":[
      {
        "type":"struct",
        "fields":[
          {
            "type":"string",
            "optional":true,
            "field":"col1"
          },
          {
            "type":"string",
            "optional":true,
            "field":"col2"
          }
        ],
        "optional":true,
        "field":"before"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"string",
            "optional":true,
            "field":"col1"
          },
          {
            "type":"string",
            "optional":true,
            "field":"col2"
          }
        ],
        "optional":true,
        "field":"after"
      }
    ],
    "optional":false
  },
  "payload":{
    "before": null,
    "after": {
      "col1": "1",
      "col2": "1"
    },
    "op": "c",
    "source": {
      "db": "default_namespace",
      "table": "table1"
    }
  }
}

canal-json

参考 Canal | Apache Flink，canal-json 格式会包含 old、data、type、database、table、pkNames 几个元素，但是 ts 并不会包含在其中。
一个输出的示例是：

{
    "old": null,
    "data": [
        {
            "col1": "1",
            "col2": "1"
        }
    ],
    "type": "INSERT",
    "database": "default_schema",
    "table": "table1",
    "pkNames": [
        "col1"
    ]
}

数据类型映射

Literal type： 反映数据的实际存储类型（对应 debezium schema 中的 type 字段）
Semantic type： 反映数据的逻辑类型（对应对应 debezium schema 中的 name 字段）。

CDC type	JSON type	Literal type	Semantic type	NOTE
TINYINT	TINYINT	INT16
SMALLINT	SMALLINT	INT16
INT	INT	INT32
BIGINT	BIGINT	INT64
FLOAT	FLOAT	FLOAT
DOUBLE	DOUBLE	DOUBLE
DECIMAL(p, s)	DECIMAL(p, s)	BYTES	org.apache.kafka.connect.data.Decimal
BOOLEAN	BOOLEAN	BOOLEAN
DATE	DATE	io.debezium.time.Date
TIMESTAMP(p)	TIMESTAMP(p)	INT64	p <=3 io.debezium.time.Timestamp p >3 io.debezium.time.MicroTimestamp
TIMESTAMP_LTZ	TIMESTAMP_LTZ	STRING	io.debezium.time.ZonedTimestamp
CHAR(n)	CHAR(n)	STRING
VARCHAR(n)	VARCHAR(n)	STRING