如何处理 kafka 主题中的旧数据？答案

【问题标题】：How can I handle old data in the kafka topic?如何处理 kafka 主题中的旧数据？
【发布时间】：2019-04-28 05:42:51
【问题描述】：

我开始使用 spark 结构化流。

我从 kafka 主题获得 readStream (startOffset: latest) 带水印，按事件时间和窗口持续时间分组，并写信给 kafka 主题。

我的问题是，如何在 spark 结构化流作业之前处理写入 kafka 主题的数据？

一开始我尝试使用 `startOffset: early' 运行。但是kafka topic中的数据太大了，所以因为yarn timeout而没有启动spark streaming进程。（即使我增加了超时值）

1。如果我只是创建一个批处理作业并按特定数据范围进行过滤。结果没有反映在火花流的当前状态中，结果的一致性和准确性似乎存在问题。

如何处理旧数据和大数据？帮帮我。

【问题讨论】：

标签： apache-spark spark-structured-streaming

【解决方案1】：

您可以尝试使用Kafka + Structured Streaming 的参数maxOffsetsPerTrigger 来接收来自Kafka 的旧数据。将此参数的值设置为您希望一次从 Kafka 接收的记录数。

用途：

sparkSession.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("subscribe", "test-name")
      .option("startingOffsets", "earliest")
      .option("maxOffsetsPerTrigger", 1)
      .option("group.id", "2")
      .option("auto.offset.reset", "earliest")
      .load()

【讨论】：