Azure / Azure/Azure-DataFactory
Mapping Data Flow / CDC support for writing to Delta Tables with Delta Lake Change Feed enabled
- 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
Assessment
This issue has not been assessed yet.