Spark structured streaming sinks to output is delayed

Question

The below spark structured streaming code collects data from Kafka at every 10 seconds:

window($"timestamp", "10 seconds")

I was expecting the results to be printed on the console every 10 seconds. But, I notice the sink to the console is happening at every ~2 mins or above. May I know what am I doing wrong?

def streaming(): Unit = {
    System.setProperty("hadoop.home.dir", "/Documents/ ")
    val conf: SparkConf = new SparkConf().setAppName("Histogram").setMaster("local[8]")
    conf.set("spark.eventLog.enabled", "false");
    val sc: SparkContext = new SparkContext(conf)
    val sqlcontext = new SQLContext(sc)
    val spark = SparkSession.builder().config(conf).getOrCreate()

    import sqlcontext.implicits._
    import org.apache.spark.sql.functions.window

    val inputDf = spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("subscribe", "wonderful")
      .option("startingOffsets", "latest")
      .load()
    import scala.concurrent.duration._

    val personJsonDf = inputDf.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "timestamp")
      .withWatermark("timestamp", "500 milliseconds")
      .groupBy(
    window($"timestamp", "10 seconds")).count()

    val consoleOutput = personJsonDf.writeStream
      .outputMode("complete")
      .format("console")
      .option("truncate", "false")
      .outputMode(OutputMode.Update())
      .start()
    consoleOutput.awaitTermination()
  }

object SparkExecutor {
  val spE: SparkExecutor = new SparkExecutor();
  def main(args: Array[String]): Unit = {
    println("test")
    spE.streaming
  }
}

Victor Victor · Accepted Answer · 2019-07-17T09:24:34

I think that you might be missing the trigger definition for querying personJsonDf during the writeStreamoperation. The 2 minute period might be a default one (not sure).

The groupBy window that you have defined, will be used in the query but it does not define its periodicity.

One way to configure this could be:

val consoleOutput = personJsonDf.writeStream
  .outputMode("complete")
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .format("console")
  .option("truncate", "false")
  .outputMode(OutputMode.Update())
  .start()

Finally, the class Trigger contains some useful methods you wanna check out.

Hope it helps.

Spark structured streaming sinks to output is delayed

1 Answers