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?