Reading highest version of delta lake table

I'm having batch jobs orchiestrated by airflow that runs spark transoformations on selected delta tables. Currently i'm processing the whole table and i want to switch to change data capture flow ultilizing delta cdf (i'm processing Create/Update/Delete)

My idea was to:

1. Read latest version of delta table
2. Process the data
3. Save the latest version on the bucket to use it in the future runs

So in upcoming runs i will first read latest version, then read latest version from previous run from the storage and use cdf to query only the changes that was introduced since the previous run.

To make read process atomic, i'm using following code snippet:

deltaTable = DeltaTable.forPath(spark, pathToTable)

# Reading latest version to be able to save it on gcp later
latest_table_version = deltaTable.history(1).collect()[0]['version']

previous_table_version = my_read_previous_table_version_from_gcp_fcn()

# Setting endingVersion to latest verison in case any changes are intoduced between those lines
df = (
   spark.read.format("delta")
   .option("readChangeFeed", "true")
   .option("startingVersion", previous_table_version )
   .option("endingVersion", latest_table_version )
   .load(pathToTable)
)

output_df = df.transform(my_transform_fcn)
save_output_fcn(output_df)

overwrite_previous_table_version_on_gcp(value=previous_table_version)

It works but feels little bit hackish. Am I using the correct approach?

Yes, it's really not the best approach to do that. It may depend on the logic of transformations you're doing, but you're really re-inventing a functionality that is by the Spark Structured Streaming out of the box - it will track the latest processed version(s) and will reprocess what is new.

Delta Lake has native support for structured streaming, so you can greatly simplify the code, you just need to make sure that you keep a checkpoint somewhere, so Spark will be able to remember what versions were processed already.

df = (
   spark.readStream.format("delta")
   .load(pathToTable)
)

output_df = df.transform(my_transform_fcn)
stream = (output_df.writeStream.format("delta")
  .option("checkpointLocation", "path-to-checkpoint")
  .trigger(availableNow=True)
  .start(outputPath")
  )
stream.awaitTermination()

.trigger(availableNow=True) will start the stream process all new data, and stop it. Otherwise it will run continuously.

By default it will reprocess all data from the beginning, but you can specify the startingVersion option in the readStream if you want to keep data that you already processed.