update date for a Dataframe and join with Kafka stream data live in spark

Viewed 20

I have a Kafka Stream source and a map table, which I want to join and then write the data to another Kafka topic. this job runs 24/7 without stop.

My issue is that the Map table that I wish to join is partitioned on date and every day I need the new updated Map table for the join.

But when the code runs it keeps using the same old map table, day after day without updating it.

import java.text.SimpleDateFormat

object joiningDF{

def newDate: String = {
val dFormat = new SimpleDateFormat("yyyy-MM-dd")
dateFormat.format(System.currentTimeMillis)
 }    

def main(args: Array[String]): Unit = {
var date=newDate
val source =spark.readStream.
      format("kafka").
      option("kafka.bootstrap.servers", "....").
      option("subscribe", "....").
      option("startingOffsets", "latest").
      load()

// MAP TABLE date variable is used to get new date daily
var map=spark.read.parquet("path/day="+date)

val joinDF=source.join(map,Seq("id"),"left")

val outQ = joinDF.
  writeStream.
  outputMode("append").
  format("kafka").
  option("kafka.bootstrap.servers", "...").
  option("topic", "...").
  option("checkpointLocation", "...").
  trigger(Trigger.ProcessingTime("300 seconds")).
  start()

  outQ.awaitTermination()
}
}

Is there a way to resolve it or a work around?

1 Answers

You could use a FileSystem source in continuous processing mode that is watching a directory into which you atomically move new versions of the file when they are ready. This will give you an updating stream to join with.

Related