时间戳

首页 > 解决方案 > 如何在结构化流中正确使用 foreachBatch.batchDF.unpersist()？（继续出错）

apache-spark - 如何在结构化流中正确使用 foreachBatch.batchDF.unpersist()？（继续出错）

问题描述

我正在使用 Spark 3.0 的结构化流。

我想要做的是将数据写入多个接收器。我需要在 Kafka 中编写一些 DataFrame 以在另一个进程中使用，并且还需要将相同的 DataFrame 存储在 Cassandra 中以供以后使用（一些仪表板等）。

对于定位过程，我编写了如下代码。我从这里参考了官方文档。

 merged_stream.writeStream
      //.trigger(Trigger.ProcessingTime("3 seconds"))
      .foreachBatch((batchDF: DataFrame, batchId: Long) => {
        batchDF.persist()
        batchDF.write
          .format("kafka")
          .option("kafka.bootstrap.servers", brokers)
          .option("kafka.compression.type", sinkCompressionType)
          .option("topic", mergeTopic)
          .mode("append")
          .save()
        batchDF.write
          .format("org.apache.spark.sql.cassandra")
          .cassandraFormat(cassandraTable, cassandraKeyspace, cassandraCluster)
          .mode("append")
          .save()
        batchDF.unpersist() //**this is the problem!!**//
      })
      .option("checkpointLocation", checkpointDir)
      .start()
      .awaitTermination()

但是，每当我batchDF.unpersist()在 foreachBatch 的最后一部分编写时，都会发生编译错误：

[error]   (function: org.apache.spark.api.java.function.VoidFunction2[org.apache.spark.sql.Dataset[org.apache.spark.sql.Row],java.lang.Long])org.apache.spark.sql.streaming.DataStreamWriter[org.apache.spark.sql.Row] <and>
[error]   (function: (org.apache.spark.sql.Dataset[org.apache.spark.sql.Row], scala.Long) => Unit)org.apache.spark.sql.streaming.DataStreamWriter[org.apache.spark.sql.Row]
[error]  cannot be applied to ((org.apache.spark.sql.DataFrame, scala.Long) => org.apache.spark.sql.DataFrame)
[error]       .foreachBatch({(batchDF: DataFrame, batchId: Long) => {
[error]        ^
[error] one error found
[error] (Compile / compileIncremental) Compilation failed

当我删除它时batchDF.unpersist()，它可以正常工作，并且我检查了数据是否可以很好地进入 Kafka 和 Cassandra。但是，很明显，由于缓存的数据保留在内存中，它很快就出现了内存不足错误。

我也尝试过使用sparkSession.catalog.clearCache()，但它似乎没有按我的预期工作。

为什么会发生此错误，因为我的代码与文档完全相同？另外，我该如何解决？

提前致谢。

标签： apache-sparkcachingcompiler-errorsspark-structured-streaming

解决方案

Spark 一直为 Scala 和 Java 提供两种不同的方法，因为 Scala 在 Scala 2.12 之前不生成 Java lambda。

  /**
   * Applies a function `f` to all rows.
   *
   * @group action
   * @since 1.6.0
   */
  def foreach(f: T => Unit): Unit = withNewRDDExecutionId {
    rdd.foreach(f)
  }

  /**
   * (Java-specific)
   * Runs `func` on each element of this Dataset.
   *
   * @group action
   * @since 1.6.0
   */
  def foreach(func: ForeachFunction[T]): Unit = foreach(func.call(_))

这是为了方便 Java 用户，但是一旦 Spark 开始支持 Scala 2.12，这些方法就会相互冲突。

Spark 社区中进行了相关讨论，但似乎做出了保持 API 兼容性的决定。也就是说，不幸的是，您需要“严格”匹配两个方法之间的签名之一，例如，Unit在 lambda 末尾添加。

推荐阅读