Pyspark Kafka 偏移范围单位

Question

我使用 Spark 作为批处理来处理来自 kafka 的日志。 在每个周期中，我的代码应该得到任何到达 kafka 消费者的东西。但是，我想限制每个周期从 kafka 获取的数据量。假设 5 GB 或 500000 条日志行..

offsetRanges = []
def storeOffsetRanges(rdd):
    global offsetRanges
    offsetRanges = rdd.offsetRanges()
    WRITE OFFSETS TO DISK
    return rdd

while True:
    host = "localhost:9092"
    offset = OffsetRange(topic, 0, fromOffset, untilOffset)
    kafka_content = KafkaUtils.createRDD(sc, {"metadata.broker.list": host}, [offset])
    kafka_content.transform(storeOffsetRanges)
    RDD TRANSFORMATIONS..

我会将偏移量存储在内存和磁盘中，以防驱动程序发生故障。但是我怎样才能强加这些 kafka 偏移量来限制每个周期的最大数据量呢？ kafka偏移范围的单位是什么？

提前致谢！

Answer 1

Kafka 偏移量单位是消息。在每个周期中，您最多会收到来自 Kafka 的 untilOffest - fromOffset 消息。但是数据只会从一个主题分区中读取，因此如果您的主题有更多分区，那么应用程序会丢失一些日志行。

您也可以尝试spark streaming with kafka direct approach。使用这种方法，您将摆脱while True，您将使用可选的背压机制基于时间（不是固定偏移量）在微批处理中处理日志行。然后您可以省略在内存中保存偏移量（流式处理会处理它），但在驱动程序重新启动的情况下仍然需要将它们保存到磁盘（参见 KafkaUtils.createDirectStream 中的 fromOffsets）。