Spark Structured Streaming：读取kafka主题中的前N条消息作为批处理查询答案

【问题标题】：Spark Structured Streaming: read first N messages in a kafka topic as a batch querySpark Structured Streaming：读取kafka主题中的前N条消息作为批处理查询
【发布时间】：2020-07-27 22:06:50
【问题描述】：

我有一个要求，我想在触发流式作业之前对 kafka 主题进行采样（用于检查其数据质量等）。采样的参数之一可能是消息数。

我引用“http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#creating-a-kafka-source-for-batch-queries”但只找到了这些方法、“startingOffsets”和“endingOffsets”。不可能从中读取前 N 条消息，因为偏移量不必是连续的（在压缩或删除消息的情况下）。

寻求任何建议或帮助。谢谢。

【问题讨论】：

标签： apache-kafka spark-structured-streaming

【解决方案1】：

引用您的帖子，您无需构建流式应用程序。您想创建一个批处理作业来执行数据质量控制。

KafkaUtils.createRDD 方法可以帮助传递您想要从 Kafka 中提取的偏移范围。它为每个 Kafka 分区获取 offsetRanges 数组。借助这些方法，可以配置要从 Kafka 主题中提取的消息数量。

以下代码将从 Kafka 主题读取 10 条消息（每个分区 5 条消息）：

val offsetRanges = Array(
  // topic, partition, inclusive starting offset, exclusive ending offset
  OffsetRange("topic_name", 0, 0, 5),
  OffsetRange("topic_name", 1, 0, 5)
)

val rdd = KafkaUtils.createRDD[String, String](sparkContext, kafkaParams, offsetRanges, PreferConsistent)

注意：Kafka TTL 会从主题中删除消息，因此您需要仔细设置偏移量，否则您的应用程序将尝试获取不存在的消息。理想情况下是使用来自 Kafka 消费者界面的beginningOffsets 方法。

您可以在 Spark 文档中找到更多信息：https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html#creating-an-rdd

【讨论】：

谢谢亚历克斯。你答案的最后一部分正是我的问题。由于压缩或删除可能会导致偏移量发生变化，因此我不能 100% 确定要读取的起始和结束偏移量。
您可以使用beginningOffsets 来获取Kafka 中的第一条可用消息，使用endOffsets 来获取Kafka 中的最后一条可用消息（在available 下是指该消息仍在Kafka 中）。