如何解决如何在结构化流中正确使用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
。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。