Azure / Azure/Azure-DataFactory

Mapping Data Flow / CDC support for writing to Delta Tables with Delta Lake Change Feed enabled

Open
#532 1 comment 0 reactions 0 assignees View on GitHub
Dominant language
PowerShell
Stars
529
Forks
623
PR merge metrics
No merged PRs in 30d

Description

Hi,

This is a follow up to a [comment ](https://github.com/Azure/Azure-DataFactory/issues/531#issuecomment-1422863003) from issue 531

I am using your new CDC resource to stream data from my Azure SQL Server to a Delta Table. Since my downstream applications need to keep track of Inserts, Updates, and Deletes, it would be very helpful if the CDC could support writing to tables that have [Delta Lake Change Feed](https://docs.databricks.com/delta/delta-change-data-feed.html) enabled.

Currently when I convert the delta table to have this enabled, the CDC activity fails. I think this might be because the ADF is writing to the table using an old Databricks Runtime, since the docs I have linked to say that once this feature is enabled, 'you can no longer write to the table using Databricks Runtime 8.1 or below.'

Would it be possible to upgrade the way ADF is writing to delta to make it compatible with this feature?

Error message from adf when this feature is enabled:
```
{
"message": "{\"StatusCode\":\"DFExecutorUserError\",\"Message\":\"Job failed due to reason: at Sink 'SinkdatabasefeeddboPlannedActivity': Delta protocol version is too new for this version of the Databricks Runtime. Please upgrade to a newer release.\",\"Details\":\"org.apache.spark.sql.delta.actions.InvalidProtocolVersionException: Delta protocol version is too new for this version of the Databricks Runtime. Please upgrade to a newer release.\\n\\tat org.apache.spark.sql.delta.DeltaLog.protocolWrite(DeltaLog.scala:294)\\n\\tat org.apache.spark.sql.delta.OptimisticTransactionImpl$class.prepareCommit(OptimisticTransaction.scala:390)\\n\\tat org.apache.spark.sql.delta.OptimisticTransaction.prepareCommit(OptimisticTransaction.scala:80)\\n\\tat org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$commit$1.apply$mcJ$sp(OptimisticTransaction.scala:287)\\n\\tat org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$commit$1.apply(OptimisticTransaction.scala:284)\\n\\tat org.apache.spark.sql.delta.OptimisticTransactionImpl$$anonfun$commit$1.apply(OptimisticTransaction.scala:284)\\n\\tat com.microsoft.spark.telemetry.delta.SynapseLoggingShim$class.recordOperation(SynapseLoggingShim.scala:72)\\n\\tat org.apache.spark.sql.delta.OptimisticTransaction.recordOperation(OptimisticTransaction.scala:80)\\n\"}",
"failureType": "UserError",
"target": "SystemPipeline_03de3c8c736a4f059fd967faf62aedcc"
}
```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.