如何在结构化流中适当地使用foreachBatch.batchDF.uninsist()?(keep出错)

Question

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

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

对于目标定位过程，我编写了如下代码。我在这里提到了官方文件。

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开始支持Scala2.12，这些方法就会相互冲突。

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