How to query all the versions from delta lake table at once to track changes made to a specific ID

user_9902813 0
Viewed 2994

I have an employee table with salary of all the employees managed using delta lake.

I can query the table based on the version or the timestamp using the time travel feature delta lake supports like this.

SELECT *
FROM DELTA.`EMPLOYEE`
VERSION AS OF 3

But I want to know the history of all the changes done to an employee across all the versions of the delta table. Something like this

SELECT *
, timestamp -- From delta table
, version -- From delta table
FROM DELTA.`EMPLOYEE`
WHERE EMPLOYEE = 'George'
WITHIN ALL VERSIONS --Never exists but just for understanding
delta-lake apache-spark-sql apache-spark
2 Answers

It is an old question but today I stumbled on it, since I had some problem to solve. I don't think Delta (delta.io) provides a method for this because Delta revolves around Time Travel to a specific point in time rather over a period.

But if I have to get this, I guess one way will be to directly read the parquet files (ignore delta logs), which will result in all the past versions/states of the record (leave Vacuum etc. aside).

Now if the requirement is to get the exact version of each record creation (which is my requirement), use something like

dataframe.withColumn("input_file", input_file_name()) which will show the exact file name the record is coming from.

Now query the .json _delta_log transaction files which will tell us which version has added which file, something like this

>>> details = spark.read.json('/data/gcs/delta/ingest/bigtable/_delta_log/*.json')
>>> details = details.select(col('add')['path'].alias("file_path")).withColumn("version",substring(input_file_name(),-6,1)).filter("file_path is not NULL")
>>> details.show(5,100)
+-------------------------------------------------------------------+-------+
|                                                          file_path|version|
+-------------------------------------------------------------------+-------+
|part-00000-148c98cc-0db1-495e-bb67-0ba1cc4fd45e-c000.snappy.parquet|      4|
|part-00001-2caa89b7-c990-47e0-b7b0-92430b15b141-c000.snappy.parquet|      4|
|part-00002-1f900af7-d819-48e9-a048-ad22e5c7ce65-c000.snappy.parquet|      4|
|part-00003-e043f466-861b-47f0-a1cf-4b67e75a5ed2-c000.snappy.parquet|      4|
|part-00000-93cc0747-ca0b-46ef-ada4-b3fb18e48925-c000.snappy.parquet|      0|
+-------------------------------------------------------------------+-------+
only showing top 5 rows

join both of these dataframes on file_path and you will see each state/version of the record along with the delta version it was created in. My example -

parquet_table = spark.read.parquet('/data/gcs/delta/ingest/bigtable/*.parquet')

>>> parquet_table.printSchema()
root
 |-- Region: string (nullable = true)
 |-- Country: string (nullable = true)
 |-- Item_Type: string (nullable = true)
 |-- Sales_Channel: string (nullable = true)


parquet_table = parquet_table.where(col("Order_ID")==913712584).\
            withColumn("input_file",substring(input_file_name(),38,1000)).\
           select(["Order_ID","Region","Country","Sales_Channel","input_file"]).\
            orderBy("Country")
            
>>> parquet_table.join(details,parquet_table.input_file == details.file_path).select("Order_ID","Region","Country","Sales_Channel","version").orderBy("version").show(100)
+---------+------------------+-------+-------------+-------+                    
| Order_ID|            Region|Country|Sales_Channel|version|
+---------+------------------+-------+-------------+-------+
|913712584|Sub-Saharan Africa|Lesotho|       Online|      0|
|913712584|Sub-Saharan Africa|Lesotho|       Online|      0|
|913712584|Sub-Saharan Africa|Lesotho|       Online|      0|
user_56934410

On Databricks, starting with the Databricks Runtime 8.2 there is a functionality called Change Data Feed that tracks what changes were made to the table, and you can pull that feed of changes either as batch or as stream for analysis or implementing change data capture-style processing.

After change data feed is enabled on the table, you can read data using batch or stream APIs, something like this:

spark.read.format("delta") \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .table("myDeltaTable")

and you'll get all changed records with additional columns that describe what kind of changes was done (insert/update/delete), when this happened (timestamp) and in which version.

Update, September 2022: Change Data Feed is now available in the open source version as well - starting with version 2.0.0.

Alex_Ott_186270
Related