Structured Spark Streaming multiple writes

Question

I am using a data stream to be written to a kafka topic as well as hbase. For Kafka, I use a format as this:

dataset.selectExpr("id as key", "to_json(struct(*)) as value")
        .writeStream.format("kafka")
        .option("kafka.bootstrap.servers", Settings.KAFKA_URL)
        .option("topic", Settings.KAFKA_TOPIC2)
        .option("checkpointLocation", "/usr/local/Cellar/zookeepertmp")
        .outputMode(OutputMode.Complete())
        .start()

and then for Hbase, I do something like this:

  dataset.writeStream.outputMode(OutputMode.Complete())
    .foreach(new ForeachWriter[Row] {
      override def process(r: Row): Unit = {
        //my logic
      }

      override def close(errorOrNull: Throwable): Unit = {}

      override def open(partitionId: Long, version: Long): Boolean = {
        true
      }
    }).start().awaitTermination()

This writes to Hbase as expected but doesn't always write to the kafka topic. I am not sure why that is happening.

*doesn't always write to the kafka topic.* What does "Doesn't always" mean? — Yuval Itzchakov, Dec 26 '17 at 08:11
That sometimes both work fine and sometimes the kafka part doesn't work at all — gmoksh, Dec 26 '17 at 08:17
https://stackoverflow.com/questions/45331883/spark-structured-streaming-multiples-queries-are-not-running-concurrently It looks like a similar issue but I can't see a solution — gmoksh, Dec 26 '17 at 08:17
I don't see you using `awaitTermination` on the first query. — Yuval Itzchakov, Dec 26 '17 at 08:19
Is your stream replayable? Each query will consume the source once. — Yuval Itzchakov, Dec 26 '17 at 08:55
Also: https://stackoverflow.com/questions/45618489/executing-separate-streaming-queries-in-spark-structured-streaming — Yuval Itzchakov, Dec 26 '17 at 09:11
@YuvalItzchakov, if I use awaitTermination on first query, the second doesn't execute at all. I am reading from a Kafka stream, can you think of a reason why this could be happening? :( — gmoksh, Dec 27 '17 at 06:22
Let us [continue this discussion in chat](http://chat.stackoverflow.com/rooms/162015/discussion-between-gmoksh-and-yuval-itzchakov). — gmoksh, Dec 27 '17 at 06:35

score 1 · Answer 1 · edited Jul 08 '19 at 12:34

Use foreachBatch in spark:

If you want to write the output of a streaming query to multiple locations, then you can simply write the output DataFrame/Dataset multiple times. However, each attempt to write can cause the output data to be recomputed (including possible re-reading of the input data). To avoid recomputations, you should cache the output DataFrame/Dataset, write it to multiple locations, and then uncache it. Here is an outline.

streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    batchDF.persist()
    batchDF.write.format(…).save(…) // location 1
    batchDF.write.format(…).save(…) // location 2
    batchDF.unpersist()
}

refer : https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#foreachbatch — , Jul 08 '19 at 12:30

Structured Spark Streaming multiple writes

1 Answers1