I am fetching data using spark and collecting it in a delta lake

user_19006871 0
Viewed 37

Now, I am running a job every day that fetches the data from delta lake of last 7 days, as the data contains the timestamp column. I am processing this data and writing to other database. As there is possibility of late data, I run this job every day and check for new data in last 7 day. But when I do so, I also get the old data with the new data. Hence the processing now I do, is for (rewriting(old data) + late data + new data). How can I stop processing the old data.Any Ideas?

azure-data-lake delta-lake apache-spark database
1 Answers

The simplest way to do that is to consume data from Delta table as a stream instead of reading data as a batch. In this case, stream will contain only data written into Delta table since the previous invocation. After you read your data, you can use foreachBatch function to write that data into your database.

Something like this (more a pseudo-code than real code):

def my_foreach_batch(df, epoch):
  .... write df into database

df = spark.readStream.option("startingVersion", "<latest_consumed_version>") \
  .format("delta").load("delta_path")
df.writeStream.option("checkpointLocation", "checkpoint_path") \
  .trigger(once=True).foreachBatch(my_foreach_batch).save()

Here are:

  • <latest_consumed_version> - last Delta version you used before switching to streaming. You can also use startingTimestamp instead (see docs)
  • checkpoint_path - path on HDFS/S3/... that will store information about versions processed by stream, so it will know where to start next time
  • trigger(once=True) - defines that stream should start, process all available data at once, and stop. Alternatively you can use trigger(availableNow=True) that is available since Spark 3.3 - it will put less pressure on the nodes when processing data.
Alex_Ott_186270
Related