apache-spark - 如何在结构化流中适本地使用 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。