使用有界Kafka数据源时如何测量Flink的through-output与latency

核心解决思路 #

要么让Flink Kafka源识别到数据集边界，任务消费完数据后自动结束，统计的全量耗时天然无额外等待；要么在业务逻辑内自定义打点，单独统计目标数据集的处理耗时与延迟，完全绕开Kafka源的无限等待逻辑。

方案1：将Kafka源配置为有界模式（最推荐） #

Flink官方Kafka连接器原生支持有界读取配置，消费到预设的停止位置后会主动终止源读取，任务正常进入FINISHED状态，任务总耗时就是真实全量数据处理耗时，无需额外处理。

• 针对Flink 1.14+版本的KafkaSource，直接调用setBounded方法指定停止偏移量即可，示例代码如下：

// 提前获取测试Kafka Topic各分区的最大偏移量，构造停止偏移量Map
Map<TopicPartition, Long> stopOffsets = new HashMap<>();
stopOffsets.put(new TopicPartition("test_topic", 0), 10000L);
stopOffsets.put(new TopicPartition("test_topic", 1), 10000L);

KafkaSource<String> boundedKafkaSource = KafkaSource.<String>builder()
    .setBootstrapServers("your_brokers:9092")
    .setTopics("test_topic")
    .setGroupId("perf_test_group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    // 配置为有界模式，消费到指定偏移量后自动停止
    .setBounded(OffsetsInitializer.offsets(stopOffsets))
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

• 针对旧版本的FlinkKafkaConsumer，关闭分区自动发现（配置flink.partition-discovery.interval-millis为-1），同时自定义实现KafkaDeserializationSchema，当消费到预设偏移量/总条数时返回isEndOfStream = true，主动终止源读取。

方案2：自定义埋点统计（无需修改源配置） #

如果不想改动现有Kafka源的配置，可以在任务内部加埋点，单独统计目标数据集的指标，不受后续等待阶段影响：

• 耗时统计：在Source算子后新增全局累加器做计数，或者用KeyedState记录每个分区的消费偏移量，当总消费条数等于测试数据集大小/所有分区都消费到预设偏移量时，记录当前时间戳作为处理结束时间，和任务启动时间的差值就是真实处理耗时，直接用于吞吐量计算。
• 延迟统计：在Source算子处给每条数据打入站时间戳，在Sink算子处计算单条数据处理耗时，只保留属于测试数据集范围内的数据的延迟结果，过滤后续空跑阶段的无效指标即可。

注意事项 #

• 测试前提前确认Kafka Topic内测试数据集的各分区最大偏移量，避免出现漏消费或者提前停止的问题
• 如果使用累加器做计数，建议把埋点放在最靠近Source的位置，避免算子链打断导致统计误差
• 不要直接使用Flink UI的整体作业延迟指标，要针对测试数据集范围做过滤，避免无效空跑数据拉低平均延迟结果

内容的提问来源于stack exchange，提问作者Dilibaba