How do I extract the accumulator value in driver?

Question

Following is how my code sequenced,

//Accumulator initialized 
val count = new LongAccumulator
sparksession.sparkContext.register(count,"count accumulator")

// Streaming Transformation
val DF = fromKafkaDF.map{
  count.add(1)
  println(count.value)  // This value is one
  //some transformation
}.writeStream.outputMode("update").format("console").start()

//trying to access the value of accumulator from driver
println(count.value)  //this value is zero

Why the value of accumulator is zero in driver? I have other logic to work on based on this accumulator. Please suggest.

My guess is that `println(count.value)` at the end is executed before transformation is executed. Because since you call `start()` the execution will start in background. To check this - try to add sleep for few seconds before. — Vladislav Varslavans, Jun 05 '18 at 09:34
I did added Thread.sleep, still the same result. value of count is still zero — prady, Jun 05 '18 at 10:28
yes it does, the counter inside the map operation is working fine and it shows the valid count. — prady, Jun 05 '18 at 10:47
Btw, you could try to use accumulator same as [here](https://spark.apache.org/docs/2.3.0/streaming-programming-guide.html#accumulators-broadcast-variables-and-checkpoints) — Vladislav Varslavans, Jun 05 '18 at 11:20

score 0 · Answer 1 · answered Sep 03 '18 at 08:23

0

For you to accumulate any value in counter, you should perform an action and check, accumulator wont be in use without any action, please check the link accumulator explained

answered Sep 03 '18 at 08:23

Harshit Saklecha

90
1
10

score 0 · Answer 2 · answered Sep 14 '19 at 08:25

use it in listeners like,

class TestListner(acc: LongAccumulator) extends StreamingQueryListener {
  override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {
    println("onQueryStarted   :" + event.toString)
  }

  override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {
    println(acc)
    acc.reset()
    println("onQueryProgress    :" + event.progress)
  }

  override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = {
    println("onQueryProgress    :" + event)
  }
}

and addListener in your main application,

    spark.streams.addListener(new TestListner(acc))

How do I extract the accumulator value in driver?

2 Answers2