I've a requirement to stream events from Azure EventHub(EH) to blob on a daily basis. I've used Spark structured streaming with Trigger.once() option along with spark check-pointing to achieve this, this work absolutely fine. But what i need further is when I run the Spark job with Trigger.once() option on a daily basis i would need to create a folder("YYYY-MM-DD") in blob on daily basis and dump all the events/data that was available in EH for that particular day.
To achieve this i was using the column='enqueuedTime' that is available in EH schema as show below and used partitionBy while writing the data to blob.
withColumn("year", year(to_date($"enqueuedTime", "MM/dd/yyyy"))).withColumn("month", month(to_date($"enqueuedTime", "MM/dd/yyyy")))
batchSignalsFinalDF
.writeStream
.format("JSON")
.partitionBy("year", "month", "day")
.option("checkpointLocation", "PATH IN BLOB")
.outputMode(OutputMode.Append).trigger(Trigger.Once())
.start("PATH IN BLOB")
This seems to be not working for one of my use case, is there a better way to achieve this!, Any suggestion would be appreciated.