Kafka DataStream连接器

Kafka Source提供了构建类来创建Kafka Source的实例。我们将通过以下示例代码介绍如何构建Kafka Source来消费input-topic最早位点的数据，消费组名称为my-group，并将Kafka消息体反序列化为字符串。

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers(brokers)
    .setTopics("input-topic")
    .setGroupId("my-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-kafka</artifactId>
    <version>${vvr-version}</version>
</dependency>

在使用Kafka DataStream连接器时，需要了解以下Kafka属性：

• 起始消费位点

  Kafka Source能够通过位点初始化器（OffsetsInitializer）来指定从不同的偏移量开始消费。内置的位点初始化器包括以下内容。

  位点初始化器	代码设置
  从最早位点开始消费。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
  从最末尾位点开始消费。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
  从时间戳大于等于指定时间的数据开始消费，单位为毫秒。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
  从消费组提交的位点开始消费，如果提交位点不存在，使用最早位点。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
  从消费组提交的位点开始消费，不指定位点重置策略。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

  说明

  • 如果以上内置的初始化器不能满足需求，也可以自己实现自定义的位点初始化器。

  • 如果未指定位点初始化器，则默认使用OffsetsInitializer.earliest()，即最早位点。

• 流模式和批模式

  Kafka Source支持流式和批式两种运行模式。默认情况下，Kafka Source设置为以流模式运行，因此作业永远不会停止，直到Flink作业失败或被取消。如果要配置Kafka Source在批模式下运行，可以使用setBounded(OffsetsInitializer)指定停止偏移量，当所有分区都达到其停止偏移量时，Kafka Source会退出运行。

  说明

  通常，流模式下Kafka Source没有停止偏移量。为了方便对代码进行调试，流模式下可以使用 setUnbounded(OffsetsInitializer) 指定停止偏移量。请留意指定流模式和批模式停止偏移量的方法名（setUnbounded 和 setBounded）是不同的。

• 动态分区检查

  为了在不重启Flink作业的情况下，处理Topic扩容或新建Topic等场景，可以在提供的Topic或Partition订阅模式下，启用动态分区检查功能。DataStream连接器中，该功能默认关闭，需要手动显式配置开启：

  KafkaSource.builder()
      .setProperty("partition.discovery.interval.ms", "10000") // 每10秒检查一次新分区。

  重要

  动态分区检查功能依赖于Kafka集群的元信息更新机制。如果Kafka集群未及时更新分区信息，可能导致新分区无法被发现。请确保Kafka集群的partition.discovery.interval.ms配置与实际情况匹配。

• 事件时间和水印

  Kafka Source默认使用Kafka消息中的时间戳作为事件时间。可以自定义水印策略（Watermark Strategy）以从消息中提取事件时间，并向下游发送水印。

  env.fromSource(kafkaSource, new CustomWatermarkStrategy(), "Kafka Source With Custom Watermark Strategy")

  如果需要了解自定义水印策略（Watermark Strategy），请参见Generating Watermarks。

  说明

  若并行Source的部分任务长期处于空闲状态（如某个Kafka分区长时间无数据输入，或Source并发数超过Kafka分区数量），可能导致Watermark生成机制失效。此时系统将无法正常触发窗口计算，造成数据处理流程停滞。

  为解决此类问题，可通过以下方式调整：

  • 配置Watermark超时机制：启用table.exec.source.idle-timeout参数，强制系统在指定超时时间后生成Watermark，确保窗口计算周期的推进。

  • 数据源优化：建议保持Kafka分区数与Source并发度的合理比例（推荐分区数≥Source并行度）。

• 消费位点提交

  Kafka Source在Checkpoint完成时，提交当前的消费位点，以保证Flink的Checkpoint状态和Kafka Broker上的提交位点一致。如果未开启Checkpoint，Kafka Source依赖于Kafka Consumer内部的位点定时自动提交逻辑，自动提交功能由enable.auto.commit和 auto.commit.interval.ms两个Kafka Consumer配置项进行配置。

  说明

  Kafka Source不依赖于Broker上提交的位点来恢复失败的作业。提交位点只是为了上报Kafka Consumer和消费组的消费进度，以在Broker端进行监控。

• 其他属性

  除了上述属性之外，还可以使用setProperties(Properties) 和setProperty(String, String) 为Kafka Source和Kafka Consumer设置任意属性。KafkaSource通常有以下配置项。

  配置项	说明
  client.id.prefix	指定用于Kafka Consumer的客户端ID前缀。
  partition.discovery.interval.ms	定义Kafka Source检查新分区的时间间隔。 说明 partition.discovery.interval.ms会在批模式下被覆盖为-1。
  register.consumer.metrics	指定是否在Flink中注册Kafka Consumer的指标。
  其他Kafka Consumer配置	Kafka Consumer的配置详情，请参见Apache Kafka。 重要 Kafka Connector会强制覆盖部分手动配置的参数项，覆盖详情如下： key.deserializer始终被覆盖为ByteArrayDeserializer。 value.deserializer始终被覆盖为ByteArrayDeserializer。 auto.offset.reset.strategy被覆盖为OffsetsInitializer#getAutoOffsetResetStrategy()。

  以下示例展示如何配置Kafka Consumer，以使用PLAIN作为SASL机制并提供JAAS配置。

  KafkaSource.builder()
      .setProperty("sasl.mechanism", "PLAIN")
      .setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";")

• 监控

  Kafka Source在Flink中注册指标，用于监控和诊断。

  • 指标范围

    Kafka source reader的所有指标都注册在KafkaSourceReader指标组下，KafkaSourceReader是operator指标组的子组。与特定主题分区相关的指标注册在KafkaSourceReader.topic.<topic_name>.partition.<partition_id>指标组中。

    例如Topic "my-topic"、分区1的当前消费位点（currentOffset）注册在<some_parent_groups>.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset。成功提交位点的次数（commitsSucceeded）注册在<some_parent_groups>.operator.KafkaSourceReader.commitsSucceeded。

  • 指标列表

    指标名称	描述	范围
    currentOffset	当前消费位点	TopicPartition
    committedOffset	当前提交位点	TopicPartition
    commitsSucceeded	成功提交的次数	KafkaSourceReader
    commitsFailed	失败的提交次数	KafkaSourceReader

  • Kafka Consumer指标

    Kafka Consumer的指标注册在指标组KafkaSourceReader.KafkaConsumer。例如Kafka Consumer指标records-consumed-total注册在<some_parent_groups>.operator.KafkaSourceReader.KafkaConsumer.records-consumed-total。

    可以使用配置项register.consumer.metrics配置是否注册Kafka消费者的指标。默认此选项设置为true。对于Kafka Consumer的指标，可参见Apache Kafka。

Flink Kafka Sink可以实现将流数据写入一个或多个Kafka Topic。

DataStream<String> stream = ...


Properties properties = new Properties();
properties.setProperty("bootstrap.servers", );
KafkaSink<String> kafkaSink =
                KafkaSink.<String>builder()
                        .setKafkaProducerConfig(kafkaProperties) // // producer config
                        .setRecordSerializer(
                                KafkaRecordSerializationSchema.builder()
                                        .setTopic("my-topic") // target topic
                                        .setKafkaValueSerializer(StringSerializer.class) // serialization schema
                                        .build())
                        .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // fault-tolerance
                        .build();

stream.sinkTo(kafkaSink);

需要配置以下参数。