如何在结构化流媒体中适当地使用 foreachBatch.batchDF.unpersist() ？（保持错误）

Question

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

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

针对定位过程，我编写了如下代码。我参考了 here.

的官方文档

 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()

但是，每当我在foreachBatch的最后部分写batchDF.unpersist()时，就会出现编译错误：

[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()，但它似乎没有按我的预期工作。

我的代码和文档完全一样，为什么会出现这个错误？另外，我该如何解决？

提前致谢。

Answer 1

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 兼容性。也就是说，不幸的是，您需要“严格”匹配两种方法之间的签名之一，例如，在 lambda 的末尾添加 Unit。

如何在结构化流媒体中适当地使用 foreachBatch.batchDF.unpersist() ？（保持错误）

how to use foreachBatch.batchDF.unpersist() appropriately in structured streaming? (keep got an error)

caching

compiler-errors

apache-spark

spark-structured-streaming

如何在结构化流媒体中适当地使用 foreachBatch.batchDF.unpersist() ？ （保持错误）

how to use foreachBatch.batchDF.unpersist() appropriately in structured streaming? (keep got an error)

caching

compiler-errors

apache-spark

spark-structured-streaming

如何在结构化流媒体中适当地使用 foreachBatch.batchDF.unpersist() ？（保持错误）