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

user_14185259 0
Viewed 1179

I'm using structured streaming from Spark 3.0.

What I'm trying to do is writing data to multiple sinks. I need to write some DataFrame in Kafka to use in another process, and also need to store same one in Cassandra for later use(some dashboard and etc).

For the targeting process, I wrote a code like below. I referred to the official doc from 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()

However, whenever I write batchDF.unpersist() in the last part of foreachBatch, it occurs compile error:

[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

When I remove the batchDF.unpersist(), it works normally and I checked that the data goes well into both Kafka and Cassandra. However, obviously, it soon got out of memory error since the cached data remains in memory.

I also tried to use sparkSession.catalog.clearCache(), but it seems not working as I intended.

Why this error is happening since my code is exactly the same as a document? Also, how can I fix it?

Thanks in advance.

spark-structured-streaming apache-spark compiler-errors caching
1 Answers

Spark has been providing two different methods for both Scala and Java, because Scala doesn't generate Java lambda before Scala 2.12.

  /**
   * 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(_))

That was for convenience to the Java users, but once Spark starts supporting Scala 2.12 these methods conflict each other.

There was relevant discussion in Spark community but the decision looked to be made to keep API compatibility. That said, unfortunately, you need to "strictly" match one of the signature between two methods, for example, add Unit at the end of lambda.

user_34801220
Related