Restart streaming query without stopping application

user_1065358 0
Viewed 1457

Im trying to restart streaming query in spark using below code inplace of query.awaitTermination(),below code will be inside an infinite loop and looks for trigger to restart query and then executes below code.Basically im trying to refresh cached df.

 query.processAllavaialble()
    query.stop()
   //oldDF is a cached Dataframe created from GlobalTempView which is of size 150GB.
          oldDF.unpersist()
    val inputDf: DataFrame = readFile(spec, sparkSession) //read file from S3
    or anyother source
    val recreateddf = inputDf.persist()
    //Start the query// here should i start query again by invoking readStream ?

But when i looked into spark documentation it says

void processAllAvailable() ///documentation says This method is intended for testing/// Blocks until all available data in the source has been processed and committed to the sink. This method is intended for testing. Note that in the case of continually arriving data, this method may block forever. Additionally, this method is only guaranteed to block until data that has been synchronously appended data to a Source prior to invocation. (i.e. getOffset must immediately reflect the addition).


stop() Stops the execution of this query if it is running. This method blocks until the threads performing execution has stopped.

So whats the better way to restart query without stopping my spark streaming application

spark-streaming apache-spark
2 Answers

This has worked for me.

Below is the scenario which I followed in spark 2.4.5 for left outer join and left join.Below process is pushing spark to read latest dimension data changes.

Process is for Stream Join with batch dimension (always update)

Step 1:-

Before starting Spark streaming job:- Make sure dimension batch data folder has only one file and the file should have at-least one record (for some reason placing empty file is not working)/

Step 2:- Start your streaming job and add a stream record in kafka stream

Step 3:- Overwrite dim data with values (the file should be same name don't change and the dimension folder should have only one file) Note:- don't use spark to write to this folder use Java or Scala filesystem.io to overwrite the file or bash delete the file and replace with new data file with same name.

Step 4:- In next batch spark is able to read updated dimension data while joining with kafka stream...

Sample Code:-

package com.databroccoli.streaming.streamjoinupdate

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.types.{StringType, StructField, StructType, TimestampType}
import org.apache.spark.sql.{DataFrame, SparkSession}

object BroadCastStreamJoin3 {

  def main(args: Array[String]): Unit = {
    @transient lazy val logger: Logger = Logger.getLogger(getClass.getName)

    Logger.getLogger("akka").setLevel(Level.WARN)
    Logger.getLogger("org").setLevel(Level.ERROR)
    Logger.getLogger("com.amazonaws").setLevel(Level.ERROR)
    Logger.getLogger("com.amazon.ws").setLevel(Level.ERROR)
    Logger.getLogger("io.netty").setLevel(Level.ERROR)

    val spark = SparkSession
      .builder()
      .master("local")
      .getOrCreate()

    val schemaUntyped1 = StructType(
      Array(
        StructField("id", StringType),
        StructField("customrid", StringType),
        StructField("customername", StringType),
        StructField("countrycode", StringType),
        StructField("timestamp_column_fin_1", TimestampType)
      ))

    val schemaUntyped2 = StructType(
      Array(
        StructField("id", StringType),
        StructField("countrycode", StringType),
        StructField("countryname", StringType),
        StructField("timestamp_column_fin_2", TimestampType)
      ))

    val factDf1 = spark.readStream
      .schema(schemaUntyped1)
      .option("header", "true")
      .csv("src/main/resources/broadcasttest/fact")


    val dimDf3 = spark.read
      .schema(schemaUntyped2)
      .option("header", "true")
      .csv("src/main/resources/broadcasttest/dimension")
      .withColumnRenamed("id", "id_2")
      .withColumnRenamed("countrycode", "countrycode_2")

    import spark.implicits._

    factDf1
      .join(
        dimDf3,
        $"countrycode_2" <=> $"countrycode",
        "inner"
      )
      .writeStream
      .format("console")
      .outputMode("append")
      .start()
      .awaitTermination

  }
}
user_40358610
Related