Deltalake change data feed on synapse notebook

@SaiSekhar, MahasivaRavi (Philadelphia)

It seems you're trying to enable the change data feed (CDF) for an existing Delta table in Azure Synapse but running into an issue where the necessary change tracking columns (_change_type, _commit_version, _commit_timestamp) aren't appearing. Let's walk through the steps to ensure it's set up correctly.

Steps to Enable Change Data Feed on an Existing Delta Table:

1. Before enabling CDF, double-check that your Delta table doesn't have any columns that conflict with the required CDF metadata columns. If your table has columns named _change_type, _commit_version, or _commit_timestamp, you'll need to rename them.
2. You’ve correctly set the property using the ALTER TABLE command, so just ensure that it has executed successfully. Here’s a reminder of the command:

      spark.sql("ALTER TABLE delta.`abfss://******@storage.dfs.core.windows.net/test/incremental_test/` SET TBLPROPERTIES (delta.enableChangeDataFeed = true)")
   

3. After running the ALTER command, you can check if the properties were set correctly by executing:

   DESCRIBE TABLE delta.`abfss://******@storage.dfs.core.windows.net/test/incremental_test/`;

Look for delta.enableChangeDataFeed in the output.

To read changes without specifying a startingVersion or startingTimestamp, you should wrap your read command in a stream as follows:

df_changes = spark.readStream.format("delta") \
 .option("readChangeFeed", "true") \
 .load("abfss://******@storage.dfs.core.windows.net/test/incremental_test/")

I have created a delta table from my side with dummy data

from pyspark.sql import SparkSession

# Initialize Spark session
spark = SparkSession.builder \
    .appName("DeltaTableExample") \
    .getOrCreate()

# Create a DataFrame
data = [("link", "zelda"), ("king k rool", "donkey kong"), ("samus", "metroid")]
columns = ["character", "franchise"]
df = spark.createDataFrame(data, columns)

# Write the DataFrame to a Delta table"my_delta_table"
df.write.format("delta").saveAsTable("my_delta_table")

I have enabled change data feed on that created table.and verified the status

spark.sql("ALTER TABLE my_delta_table SET TBLPROPERTIES (delta.enableChangeDataFeed = true)")

I have read the change data feed for the table

changes_df = spark.read.format("delta").option("readChangeData", "true") \
    .option("startingVersion", 1).option("endingVersion", 5).table("my_delta_table")
changes_df.show()

Hope this helps. Do let us know if you any further queries.

Hi @phemanth,
Thanks for the update,

we are looking to change delta as file not table and can you please try to read the datachangefeed without passing starting version in the pyspark syntax and can you try to enable CDF on delta as a file, please share the code of it.

Thanks in Advance

Thank you!