结构化流媒体OOM答案

【问题标题】：Structured Streaming OOM结构化流媒体OOM
【发布时间】：2020-09-09 02:42:48
【问题描述】：

我在 k8s 运算符上部署了一个结构化流作业，它只是从 kafka 读取、反序列化、添加 2 列并将结果存储在数据湖中（尝试了 delta 和 parquet），几天后执行器增加了内存，最终我得到OOM。输入记录的 kbs 非常低。 P.s 我使用完全相同的代码，但使用 cassandra 作为接收器，现在运行了将近一个月，没有任何问题。有什么想法吗？

enter image description here

我的代码

spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", MetisStreamsConfig.bootstrapServers)
    .option("subscribe", MetisStreamsConfig.topics.head)
    .option("startingOffsets", startingOffsets)
    .option("maxOffsetsPerTrigger", MetisStreamsConfig.maxOffsetsPerTrigger)
    .load()
    .selectExpr("CAST(value AS STRING)")
    .as[String]
    .withColumn("payload", from_json($"value", schema))

    // selection + filtering
    .select("payload.*")
    .select($"vesselQuantity.qid" as "qid", $"vesselQuantity.vesselId" as "vessel_id", explode($"measurements"))
    .select($"qid", $"vessel_id", $"col.*")
    .filter($"timestamp".isNotNull)
    .filter($"qid".isNotNull and !($"qid"===""))
    .withColumn("ingestion_time", current_timestamp())
    .withColumn("mapping", MappingUDF($"qid"))
  writeStream
    .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
      log.info(s"Storing batch with id: `$batchId`")
      val calendarInstance = Calendar.getInstance()

      val year = calendarInstance.get(Calendar.YEAR)
      val month = calendarInstance.get(Calendar.MONTH) + 1
      val day = calendarInstance.get(Calendar.DAY_OF_MONTH)
      batchDF.write
        .mode("append")
        .parquet(streamOutputDir + s"/$year/$month/$day")
    }
    .option("checkpointLocation", checkpointDir)
    .start()

我改成 foreachBatch 因为使用 delta 或 parquet 和 partitionBy 会导致问题更快

【问题讨论】：

可以显示代码吗？
@Srinivas 我用代码更新了帖子
我使用 cassandra 作为接收器的确切代码，但我没有问题
你能发布 spark-submit 命令吗？？
@Srinivas 使用 google-spark 运算符部署。所以我只以编程方式传递配置。 driver.memory 2g 以及 executor 和 sql.shuffle.partitions 10

标签： apache-spark apache-spark-sql spark-streaming spark-structured-streaming

【解决方案1】：

Spark 3.1.0 中解决了一个错误。

见https://github.com/apache/spark/pull/28904

解决问题的其他方法和调试功劳：

https://www.waitingforcode.com/apache-spark-structured-streaming/file-sink-out-of-memory-risk/read

即使您正在使用 foreachBatch，您也会发现这很有帮助...

【讨论】：

【解决方案2】：

对于一些使用 partitionBy 编写 Delta Lake（或 Parquet）输出的 Structured Streaming Spark 2.4.4 应用程序，我遇到了同样的问题。

似乎与容器内的 jvm 内存分配有关，正如此处彻底解释的那样： https://merikan.com/2019/04/jvm-in-a-container/

我的解决方案（但取决于您的 jvm 版本）是在我的 spark 应用程序的 yaml 定义中添加一些选项：

spec:
    javaOptions: >-
        -XX:+UnlockExperimentalVMOptions -XX:+UseCGroupMemoryLimitForHeap

这样我的 Streamin 应用程序运行正常，内存量正常（驱动程序 1GB，执行程序 2GB）

编辑：虽然第一个问题似乎已经解决（控制器杀死 pod 以消耗内存），但仍然存在非堆内存大小缓慢增长的问题；几个小时后，司机/执行者被杀......

【讨论】：