如何在结构化流中正确使用foreachBatch.batchDF.unpersist？ 保持错误

问题描述

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

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

对于定位过程，我编写了如下代码。我提到了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()，但它似乎没有达到我的预期。

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

谢谢。

解决方法

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。