Spark-Streaming 最早因 kafka 起始偏移而挂起（Kafka 2，spark 2.4.3）答案

【问题标题】：Spark-Streaming hangs with kafka starting offset at earliest (Kafka 2, spark 2.4.3)Spark-Streaming 最早因 kafka 起始偏移而挂起（Kafka 2，spark 2.4.3）
【发布时间】：2020-01-18 19:08:15
【问题描述】：

我遇到了 Spark-Streaming 和 Kafka 的问题。在运行示例程序以从 Kafka 主题消费并将微批处理结果输出到终端时，当我设置选项时，我的工作似乎挂起：

df.option("startingOffsets", "earliest")

从最新的偏移量开始工作正常，结果会在每个微批次流过时打印到终端。

我在想这可能是一个资源问题——我正在尝试从一个包含大量数据的主题中读取。但是我似乎没有内存/cpu 问题（使用本地 [*] 集群运行此作业）。这项工作似乎从未真正开始，而只是悬而未决：

19/09/17 15:21:37 INFO Metadata: Cluster ID: JFXVL24JQ3K4CEbE-VA58A

  val sc = new SparkConf().setMaster("local[*]").setAppName("spark-test")
  val streamContext = new StreamingContext(sc, Seconds(1))
  val spark = SparkSession.builder().appName("spark-test")
    .getOrCreate()

  val topic = "topic.with.alotta.data"

  //subscribe tokafka
  val df = spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "127.0.0.1:9092")
    .option("subscribe", topic)
    .option("startingOffsets", "earliest")
    .load()

 //write
 df.writeStream
    .outputMode("append")
    .format("console")
    .option("truncate", "false")
    .start()
    .awaitTermination()

我希望看到打印到控制台的结果....但是，正如我所提到的，应用程序似乎挂起。有什么想法吗？感觉就像一个火花资源问题（因为我正在针对一个有大量数据的主题运行一个本地“集群”。我是否缺少关于流数据帧的性质的一些东西？

【问题讨论】：

我在 Spark 2.4.4 中看到了同样的情况，但最终它在 java.lang.OutOfMemoryError: Java heap space 中消失了

标签： apache-spark apache-kafka kafka-consumer-api spark-structured-streaming

【解决方案1】：

写入控制台会导致每次触发时将所有数据收集到驱动程序的内存中。由于您当前没有限制批次的大小，这意味着整个主题内容正在累积在驱动程序中。见https://spark.apache.org/docs/2.4.3/structured-streaming-programming-guide.html#output-sinks

对批量大小设置限制应该可以解决您的问题。从 Kafka 读取时尝试添加 maxOffsetsPerTrigger 设置...

  val df = spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "127.0.0.1:9092")
    .option("subscribe", topic)
    .option("startingOffsets", "earliest")
    .option("maxOffsetsPerTrigger", 1000)
    .load()

详情请见https://spark.apache.org/docs/2.4.3/structured-streaming-kafka-integration.html。

【讨论】：