Flink SQL Kafka Connector

Flink 版本：1.13

Kafka Connector 提供了从 Kafka topic 中消费和写入数据的能力。

1. 依赖

无论是使用构建自动化工具（例如 Maven 或 SBT）的项目还是带有 SQL JAR 包的 SQL 客户端，如果想使用 Kafka Connector，都需要引入如下依赖项：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka_2.11</artifactId>
  <version>1.13.0</version>
</dependency>

如果是使用的 SQL 客户端，需要下载对应的 Jar 包放在 flink 安装目录的 lib 文件夹下。

2. 创建Kafka表

如下示例展示了如何创建一个 Kafka Source 表：

CREATE TABLE kafka_source_table (
  uid STRING COMMENT '用户Id',
  wid STRING COMMENT '微博Id',
  tm STRING COMMENT '发微博时间',
  content STRING COMMENT '微博内容'
) WITH (
  'connector' = 'kafka',
  'topic' = 'behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'kafka-example',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'json',
  'value.json.ignore-parse-errors' = 'true'
);

3. 获取元数据

如下 Connector 元数据可以在表定义中通过元数据列来获取：

R/W 列定义了一个元数据是可读（R）还是可写（W）。只读列必须声明为 VIRTUAL 以在 INSERT INTO 操作中排除它们。

如下示例展示了如何使用这些元数据字段：

CREATE TABLE kafka_meta_source_table (
  -- 元数据字段
  `topic` STRING METADATA VIRTUAL, -- 不指定 FROM
  `partition_id` STRING METADATA FROM 'partition' VIRTUAL, -- 指定 FROM
  `offset` BIGINT METADATA VIRTUAL,  -- 不指定 FROM
  `ts` TIMESTAMP(3) METADATA FROM 'timestamp' VIRTUAL, -- 指定 FROM
  -- 业务字段
  `uid` STRING COMMENT '用户Id',
  `wid` STRING COMMENT '微博Id',
  `tm` STRING COMMENT '发微博时间',
  `content` STRING COMMENT '微博内容'
) WITH (
  'connector' = 'kafka',
  'topic' = 'behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'kafka-meta-example',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'json',
  'value.json.ignore-parse-errors' = 'true'
);

如果表字段名称与元数据字段名称相同，可以不用指定 FROM，例如，offset。如果想使用不同的字段名称，需要指定 FROM，比如，partition_id。

完整示例代码请查阅： kafka_meta_example

4. Connector 参数

5. Key 与 Value Format

Kafka 消息 Key 和 Value 部分都可以使用指定的 Format 来序列化或反序列化。Key Format 用来序列化和反序列化 Kafka 消息的 Key 部分，Value Format 用来序列化和反序列化 Kafka 消息的 Value 部分。

5.1 只有 Value Format

由于 Kafka 消息中 Key 是可选的，因此以下语句只配置 Value Format 来读取和写入记录。’format’ 选项与 ‘value.format’ 意义相同，两个配置项选择其中一个配置即可。所有的 Format 配置使用 Format 标识符作为前缀，例如，Json Foramt 配置均以 json 作为前缀：

CREATE TABLE kafka_value_source_table (
  uid STRING COMMENT '用户Id',
  wid STRING COMMENT '微博Id',
  tm STRING COMMENT '发微博时间'
) WITH (
  'connector' = 'kafka',
  'topic' = 'behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'kafka-connector-value',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.ignore-parse-errors' = 'true',
  'json.fail-on-missing-field' = 'true'
);

5.2 Key 与 Value Format

如下示例展示了如何配置一起使用 Key 和 Value Format。Format 配置使用 ‘key’ 或 ‘value’ 再加上 Format 标识符作为前缀。对于 Json Foramt，Key Format 均以 ‘key.json’ 作为前缀，Value Format 均以 ‘value.json’ 作为前缀：

CREATE TABLE kafka_key_value_source_table (
  uid STRING COMMENT '用户Id',
  wid STRING COMMENT '微博Id',
  tm STRING COMMENT '发微博时间'
) WITH (
  'connector' = 'kafka',
  'topic' = 'behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'kafka-connector-key-value',
  'scan.startup.mode' = 'earliest-offset',
  -- Key Format
  'key.format' = 'json',
  'key.json.ignore-parse-errors' = 'true',
  'key.fields' = 'uid;wid',
  -- Value Format
  'value.format' = 'json',
  'value.json.fail-on-missing-field' = 'false',
  'value.fields-include' = 'EXCEPT_KEY'
);

完整示例代码请查阅： kafka-connector-key-value

假设 Kafka 消息 Key 数据如下所示：

{
  "uid": "fa5aed172c062c61e196eac61038a03b",
  "wid": "7cce78a4ad39a91ec1f595bcc7fb5eba"
}

Kafka 消息 Value 数据如下所示：

{
  "tm": "2015-08-01 14:06:31",
  "content": "卖水果老人"
}

在解析上述 Kafka 消息时，需要添加 ‘value.fields-include’ = ‘EXCEPT_KEY’ 参数，指定 Key 相关字段不在 Value 中，否则 uid、wid 会被当成 Value 的一部分进行解析，从而导致解析不出数据。

假设 Kafka Key 数据格式如下：

{
  "uid": "fa5aed172c062c61e196eac61038a03b",
  "wid": "7cce78a4ad39a91ec1f595bcc7fb5eba"
}

Kafka Value 数据格式如下：

{
  "uid": "fa5aed172c062c61e196eac61038a03b",
  "wid": "7cce78a4ad39a91ec1f595bcc7fb5eba",
  "tm": "2015-08-01 14:06:31",
  "content": "卖水果老人"
}

在解析上述 Kafka 消息时，使用 ‘EXCEPT_KEY’ 或者 ‘ALL’ 均可以。

5.1.3 重名字段

如果 Key Format 和 Value Format 中包含了相同名称的字段，那么 Connnector 无法根据 Schema 信息将这些列区分为 Key 字段和 Value 字段。’key.fields-prefix’ 配置项可以在表结构中为 Key 字段指定一个唯一名称，并在配置 Key Format 的时候保留原名。如下示例展示了在 Key 和 Value Format 中同时包含 version 字段的情况：

CREATE TABLE kafka_same_name_source_table (
  `key_uid` STRING COMMENT 'kafka消息的key',
  `uid` STRING COMMENT '用户Id',
  `wid` STRING COMMENT '微博Id',
  `tm` STRING COMMENT '发微博时间'
) WITH (
  'connector' = 'kafka',
  'topic' = 'behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'kafka-connector-same-name',
  'scan.startup.mode' = 'earliest-offset',
  -- Key Format
  'key.format' = 'json',
  'key.fields-prefix' = 'key_',
  'key.fields' = 'key_uid',
  'key.json.ignore-parse-errors' = 'true',
  -- Value Format
  'value.format' = 'json',
  'value.json.ignore-parse-errors' = 'true',
  'value.json.fail-on-missing-field' = 'false',
  'value.fields-include' = 'EXCEPT_KEY' --key 字段不在消息 Value 部分中
);

完整示例代码请查阅： kafka-connector-same-name

6. 特性

6.1 Topic 和 Partition 自动发现

topic 和 topic-pattern 配置项决定了 Source 消费的 topic 或 topic 的匹配规则。topic 配置项可接受使用分号间隔的 topic 列表，例如 topic-1;topic-2。topic-pattern 配置项使用正则表达式来发现可以匹配的 topic。例如 topic-pattern 设置为 test-topic-[0-9]，那么在作业启动时，与这个正则表达式相匹配的 topic（以 test-topic- 开头，以一位数字结尾）都会被消费者订阅。

为了允许消费者在作业启动之后能够自动发现创建的 topic，需要将 scan.topic-partition-discovery.interval 配置为一个非负值。这能够使消费者发现与指定模式相匹配的新 topic 中的 partition。

注意 topic 列表和 topic 匹配规则只适用于 source。对于 sink 端，Flink 目前只支持单一 topic。

6.2 起始消费位点

scan.startup.mode 配置项决定了 Kafka 消费者的启动模式。具体值如下所示：

• group-offsets：从指定的消费组提交到 Zookeeper 或者 Kafka Broker 的偏移量开始消费。
• earliest-offset：从最早的偏移量开始消费。
• latest-offset：从最末尾偏移量开始消费。
• timestamp：从用户为每个 partition 指定的时间戳开始消费。
• specific-offsets：从用户为每个 partition 指定的偏移量开始消费。

默认值为 group-offsets 表示从 Zookeeper 或者 Kafka Broker 中最近一次提交的偏移量开始消费。

如果使用了 timestamp，必须也要配置另外一个配置项 scan.startup.timestamp-millis，来指定一个从格林尼治标准时间 1970 年 1 月 1 日 00:00:00.000 开始计算的毫秒单位时间戳作为起始时间。如果使用了 specific-offsets，必须也要配置另外一个配置项 scan.startup.specific-offsets，来为每个 partition 指定起始偏移量，例如，选项值 partition:0,offset:42;partition:1,offset:300 表示 partition 0 从偏移量 42 开始，partition 1 从偏移量 300 开始。

6.3 Sink 分区

配置项 sink.partitioner 指定了从 Flink 分区到 Kafka 分区的映射关系。默认情况下，Flink 使用 Kafka 默认分区器来对消息进行分区。默认分区器对没有消息 Key 的消息使用粘性分区策略（sticky partition strategy） 进行分区，对含有消息 Key 的消息使用 murmur2 哈希算法计算分区。为了控制消息到分区的路由，也可以提供一个自定义的 Sink 分区器。’fixed’ 分区器会将相同 Flink 分区中的消息写入同一个 Kafka 分区，从而减少网络连接的开销。

6.4 一致性保证

默认情况下，如果在启用 Checkpoint 模式下执行查询，Kafka Sink 会按照 At-Least-Once 语义保证将数据写入到 Kafka Topic 中。当 Flink Checkpoint 启用时，kafka Sink 可以提 Exactly-Once 语义保证。除了启用 Flink Checkpoint，还可以通过选择不同的 sink.semantic 选项来选择三种不同的运行模式：

• None：不保证任何语义。输出的记录可能重复或者丢失。
• At-Least-Once (默认设置)：保证不会有记录丢失，但可能会重复。
• Exactly-Once：使用 Kafka 事务提供 Exactly-Once 语义。当使用事务向 Kafka 写入数据时，不要忘记设置所需的隔离级别（read_committed 或者 read_uncommitted，后者是默认值）。

6.5 数据类型映射

Kafka 将消息 Key 和值存储为字节，因此 Kafka 没有 Schema 以及数据类型。Kafka 消息按照配置 Format 进行反序列化和序列化，例如 csv、json、avro。因此，数据类型映射由特定 Format 决定。