Kafka SQL连接器

背景信息

前提条件

注意事项

目前不推荐使用事务写入，这是 Flink 社区和 Kafka 社区的设计缺陷所致。当设置sink.delivery-guarantee = exactly-once，Kafka Connector 会启用事务写入，存在三个已知问题：

• 每个 Checkpoint 会生成一个 Transaction ID。如果 Checkpoint 间隔太短，Transaction ID会过多。Kafka 集群的 Coordinator 可能因此内存不足，从而破坏 Kafka 集群的稳定性。

• 每个事务会创建一个 Producer 实例。如果同时提交的事务太多，TaskManager 的内存可能耗尽，从而破坏 Flink 作业的稳定性。

• 多个 Flink 作业若使用相同的sink.transactional-id-prefix，它们生成的事务 ID 可能冲突。一个作业写入失败时，会阻塞 Kafka 分区的 LSO（Log Start Offset）前进，这会影响所有消费者读取该分区的数据。

如果你需要 Exactly-Once 语义，改用 Upsert Kafka 写入主键表，并用主键保证幂等性。如果需要使用事务写入，请参见EXACTLY_ONCE语义注意事项。

网络连接排查

Flink作业启动时若报错Timed out waiting for a node assignment，通常是因为 Flink 与 Kafka 之间网络不通。

SQL

Kafka连接器可以在SQL作业中使用，作为源表或者结果表。

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

元信息列

CREATE TABLE kafka_source (
  --读取消息所属的topic作为`record_topic`字段
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  --读取ConsumerRecord中的时间戳作为`ts`字段
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  --读取消息的offset作为`record_offset`字段
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE kafka_sink (
  --将`ts`字段中的时间戳作为ProducerRecord的时间戳写入Kafka
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

WITH参数

• 通用

  参数	说明	数据类型	是否必填	默认值	备注
  connector	表类型。	String	是	无	固定值为kafka。
  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。 不能通过该方式修改以下配置，因为它们会被Kafka连接器覆盖： key.deserializer value.deserializer
  format	读取或写入Kafka消息value部分时使用的格式。	String	否	无	支持的格式 csv json avro debezium-json canal-json maxwell-json avro-confluent raw 说明 更多format参数设置请参见Format参数。
  key.format	读取或写入Kafka消息key部分时使用的格式。	String	否	无	支持的格式 csv json avro debezium-json canal-json maxwell-json avro-confluent raw 说明 使用该配置时，key.options配置是必填的。
  key.fields	Kafka消息key部分对应的源表或结果表字段。	String	否	无	多个字段名以分号（;）分隔。例如field1;field2
  key.fields-prefix	为所有Kafka消息key部分指定自定义前缀，以避免与消息value部分格式字段重名。	String	否	无	该配置项仅用于源表和结果表的列名区分，解析和生成Kafka消息key部分时，该前缀会被移除。 说明 使用该配置时，value.fields-include必须配置为EXCEPT_KEY。
  value.format	读取或写入Kafka消息value部分时使用的格式。	String	否	无	该配置等同于format，只能设置 format 或 value.format 中的一个。如果同时配置，value.format会覆盖format。
  value.fields-include	在解析或生成Kafka消息value部分时，是否要包含消息key部分对应的字段。	String	否	ALL	参数取值如下： ALL（默认值）：所有列都会作为Kafka消息value部分处理 EXCEPT_KEY：除去key.fields定义的字段，剩余字段作为Kafka消息value部分处理

• 源表

  参数	说明	数据类型	是否必填	默认值	备注
  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	否	KafkaSource-{源表表名}	如果指定的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的所有分区数据。 说明 在实时计算引擎VVR 6.0.x版本中，动态分区检测默认为关闭。自8.0版本起该功能默认打开，检测间隔默认设置为5分钟。
  scan.header-filter	根据Kafka数据是否包含指定的消息头（Header）对数据进行条件过滤。	String	否	无	Header key和value使用冒号（:）分隔，多个header条件之间使用逻辑运算符（&、|）连接，支持取反逻辑运算符（!）。例如depart:toy|depart:book&!env:test表示保留header中包含depart=toy或depart=book，且不包含env=test的Kafka数据。 说明 仅实时计算引擎VVR 8.0.6及以上版本支持配置该参数。 暂不支持括号运算。 逻辑运算顺序为从左至右。 Header value会以UTF-8格式转换为字符串，与参数指定的header value进行比较。
  scan.check.duplicated.group.id	是否检查通过properties.group.id指定的消费者组有重复。	Boolean	否	false	参数取值如下： true：在启动作业前，系统会检查消费者组是否存在重复。若发现重复，作业将报错并停止运行，从而避免与现有消费者组产生冲突。 false：直接启动作业，不检查消费者组冲突。 说明 仅VVR 6.0.4及以上版本支持该参数。

• 结果表

  参数	说明	数据类型	是否必填	默认值	备注
  topic	写入的topic名称。	String	是	无	无
  sink.partitioner	从Flink并发到Kafka分区的映射模式。	String	否	default	取值如下： default（默认值）：使用Kafka默认的分区模式 fixed：每个Flink并发对应一个固定的Kafka分区。 round-robin：Flink并发中的数据将被轮流分配至Kafka的各个分区。 自定义分区映射模式：如果fixed和round-robin不满足需求，可以创建一个FlinkKafkaPartitioner的子类来自定义分区映射模式。例如org.mycompany.MyPartitioner
  sink.delivery-guarantee	Kafka结果表的语义模式。	String	否	at-least-once	取值如下： none：不保证任何语义，数据可能会丢失或重复。 at-least-once（默认值）：保证数据不丢失，但可能会重复。 exactly-once：使用Kafka事务保证数据不会丢失和重复。 说明 在使用exactly-once语义时，sink.transactional-id-prefix是必填的。
  sink.transactional-id-prefix	在exactly-once语义下使用的Kafka事务ID前缀。	String	否	无	只有sink.delivery-guarantee配置为exactly-once时该配置才会生效。
  sink.parallelism	Kafka结果表算子的并发数。	Integer	否	无	上游算子的并发，由框架决定。

安全与认证

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";'
)

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_SSL',
  /*SSL配置*/
  /*配置服务端提供的truststore (CA 证书) 的路径*/
  /*文件管理上传的文件会存放在/flink/usrlib/路径下*/
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /*如果要求客户端认证，则需要配置keystore (私钥) 的路径*/
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /*客户端验证服务器地址的算法，空值表示禁用服务器地址验证*/
  'properties.ssl.endpoint.identification.algorithm' = '',
  /*SASL配置*/
  /*将SASL机制配置为as SCRAM-SHA-256*/
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /*配置JAAS*/
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

示例中提到的CA证书和私钥可使用实时计算控制台的文件管理功能上传至平台。上传后文件存放在/flink/usrlib目录下，需要使用的CA证书文件名为my-truststore.jks，则上传后在WITH参数中有两种方式可以设置'properties.ssl.truststore.location'来使用该证书：

• 配置'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks'，采用这种方式Flink运行期间不需要动态下载OSS文件，但是不支持调试模式。

• 实时计算引擎版本为 VVR 11.5 及以上，可以配置properties.ssl.truststore.location和properties.ssl.keystore.location为OSS绝对路径地址，文件路径格式为oss://flink-fullymanaged-<工作空间ID>/artifacts/namespaces/<项目空间名称>/<文件名>。采用这种方式会在Flink运行期间动态下载OSS文件，支持调试模式。

源表起始位点

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  --从最早位点开始消费
  'scan.startup.mode' = 'earliest-offset',
  --从最末尾位点开始消费
  'scan.startup.mode' = 'latest-offset',
  --从消费者组"my-group"的已提交位点开始消费
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- 如果 "my-group" 为首次使用，则从最早位点开始消费
  'properties.auto.offset.reset' = 'latest', -- 如果 "my-group" 为首次使用，则从最末尾位点开始消费
  --从指定的毫秒时间戳1655395200000开始消费
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  --从指定位点开始消费
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

源表位点提交

Kafka源表只在checkpoint成功后将当前消费位点提交至Kafka集群。如果checkpoint间隔设置较长，在Kafka集群侧观察到的消费位点会有延迟。在进行checkpoint时，Kafka源表会将当前读取进度存储在状态中，并不依赖于提交到集群上的位点进行故障恢复，提交位点仅仅是为了在Kafka侧能够监控到读取进度，位点提交失败不会对数据正确性产生任何影响。

结果表自定义分区器

如果内置的Kafka Producer分区模式无法满足需求，可以实现自定义分区模式将数据写入对应的分区。自定义分区器需要继承FlinkKafkaPartitioner，开发完成后编译JAR包，使用文件管理功能上传至实时计算控制台。上传并引用完成后，请在WITH参数中设置sink.partitioner参数，参数值为分区器完整的类路径，如org.mycompany.MyPartitioner。

Kafka、Upsert Kafka或Kafka JSON catalog的选择

Kafka是一种只能添加数据的消息队列系统，无法进行数据的更新和删除操作，因此在流式SQL计算中无法处理上游的CDC变更数据和聚合、联合等算子的回撤逻辑。如果需要将含有变更或回撤类型的数据写入Kafka，请使用对变更数据进行特殊处理的Upsert Kafka结果表。

为了方便将上游数据库中一个或多个数据表中的变更数据批量同步到Kafka中，可以使用Kafka JSON catalog。如果Kafka中存储的数据格式为JSON，使用Kafka JSON catalog可以省去定义schema和WITH参数的步骤。详情可参见管理Kafka JSON Catalog。

使用示例

CREATE TEMPORARY TABLE kafka_source (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'source',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE kafka_sink (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

INSERT INTO kafka_sink SELECT id, name, age FROM kafka_source;

CREATE TEMPORARY TABLE kafkaTable (
  `offset` INT NOT NULL METADATA,
  `part` BIGINT NOT NULL METADATA FROM 'partition',
  PRIMARY KEY (`part`, `offset`) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.infer-schema.flatten-nested-columns.enable' = 'true'
    --可选，将嵌套列全部展开。
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `key_id` INT NOT NULL,
  `val_name` VARCHAR(200)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'val_',
  'value.fields-include' = 'EXCEPT_KEY'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`(
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `distinct_id` INT NOT NULL,
  `properties` STRING,
  `timestamp` TIMESTAMP_LTZ METADATA,
  `date` AS CAST(`timestamp` AS DATE)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH (
   'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable
ADD COLUMN
  `order_id` AS COALESCE(JSON_VALUE(`properties`, '$.order_id'), 'default');
--使用COALESCE处理空值情况。

{
  "id": 101,
  "name": "VVP",
  "properties": {
    "owner": "阿里云",
    "engine": "Flink"
  }
}

CREATE TEMPORARY TABLE kafka_source (
  id          VARCHAR,
  `name`      VARCHAR,
  properties  ROW<`owner` STRING, engine STRING>
) WITH (
  'connector' = 'kafka',
  'topic' = 'xxx',
  'properties.bootstrap.servers' = 'xxx',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

EXACTLY_ONCE语义注意事项

• 配置消费者隔离级别

  所有消费 Kafka 数据的应用必须设置 isolation.level：

  • read_committed：只读取已提交的数据。

  • read_uncommitted（默认）：可读取未提交的数据。

  EXACTLY_ONCE 依赖read_committed。否则消费者可能看到未提交数据，破坏一致性。

• 事务超时与数据丢失

  Flink 从 Checkpoint 恢复时，仅依赖该 Checkpoint 开始前已提交的事务。如果作业崩溃到重启的时间超过 Kafka 事务超时，Kafka 会自动中止事务，导致数据丢失。

  • Kafka Broker 默认transaction.max.timeout.ms = 15 分钟。

  • Flink Kafka Sink 默认设置 transaction.timeout.ms = 1 小时。

  • 你必须在 Broker 端提高transaction.max.timeout.ms，使其不小于 Flink 的设置。

• Producer 池与 Checkpoint 并发

  EXACTLY_ONCE 模式使用固定大小的 Kafka Producer 池。每个 Checkpoint 占用池中的一个 Producer。如果并发 Checkpoint 数超过池大小，作业会失败。

  请根据最大并发 Checkpoint 数调整 Producer 池大小。

• 并行度缩容限制

  如果作业在第一个 Checkpoint 前失败，重启后不会保留原有 Producer 池信息。因此，在第一个 Checkpoint 完成前，不要缩减作业并行度。如必须缩容，并行度不得低于FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR。

• 事务阻塞读取

  在read_committed模式下，任何未结束（未提交也未中止）的事务会阻塞整个 Topic 的读取。

  例如：

  • 事务 1 写入数据。

  • 事务 2 写入并提交数据。

  • 只要事务 1 未结束，事务 2 的数据对消费者不可见。

  因此：

  • 正常运行时，数据可见延迟约等于 Checkpoint 间隔。

  • 作业失败时，正在写入的 Topic 会阻塞消费者，直到作业重启或事务超时，极端情况下，事务超时甚至会影响读取。

常见问题

• Flink如何获取Kafka中的JSON数据？

• 为什么Kafka源表数据基于Event Time的窗口后，不能输出数据？

• 如何连接配置了安全信息的Kafka集群？

• 如何解决字段命名冲突的问题？

• 空间管理与操作