Azure Databricks技术问询：从Delta表读取流写入ORC文件是否仅包含合并变更及方案可行性分析

Hey Tim, let's break down your two questions step by step, using Delta Lake's structured streaming capabilities as context:

Question 1: Will the generated ORC files only contain the changed data from the latest Delta merge? #

The answer depends entirely on whether you've enabled Delta Lake Change Data Feed (CDF) on your target Delta table:

• Without CDF enabled (default behavior): When using spark.readStream.format("delta").load(...), the stream treats the Delta table as an append-only source. This means it will only capture newly inserted rows from your hourly Merge operation, not the updated rows (since updates overwrite existing rows in the Delta table, and the default stream reader doesn't track row-level changes). Your ORC files would miss all updated records, only including new inserts from the Merge.
• With CDF enabled: If you turn on CDF for the Delta table (via ALTER TABLE ... SET TBLPROPERTIES (delta.enableChangeDataFeed = true)), modify your stream reader to explicitly read the change feed:

  deltatbl_event_readstream = spark.readStream.format("delta") 
      .option("readChangeFeed", "true")
      .load("/mnt/delta/myadlsaccnt/user_events")
  

  This stream will capture all change events from the Merge: both newly inserted rows (_change_type = "insert") and the updated versions of existing rows (_change_type = "update_postImage"). In this case, your ORC files will include exactly the changed data from the latest Merge (assuming your checkpoint correctly tracks processed changes).

Question 2: Are there potential issues with writing only changed/updated data to ORC files? #

Yes, several critical considerations need to be addressed to avoid data integrity gaps or processing errors:

• Missing updates without CDF: As noted above, skipping CDF means your ORC output will only include new inserts, not updates. This creates a permanent discrepancy between the up-to-date Delta table and the stale ORC files.
• Duplicate change events: If your Delta Merge job retries (e.g., due to a transient failure), the same change events may be regenerated. Your current loadToLocation function uses mode("append") without deduplication, which will result in duplicate records in the ORC storage. You’ll need to add logic like dropping duplicates on primaryKey and _commit_version (a CDF-provided field) to prevent this.
• Downstream processing complexity: If downstream systems consuming the ORC files expect a complete, up-to-date dataset (not just incremental changes), they’ll need to maintain their own copy of the data and apply changes incrementally. This adds significant complexity compared to having a full snapshot available.
• Stale partition data: If you partition ORC files by updateddate, and a record’s updateddate is modified in the Merge, the CDF will include the updated row with the new updateddate. However, your current write logic appends to the ORC location, so the old version of the row (with the original updateddate) will remain in the old partition. This leads to stale, conflicting data unless you implement a cleanup process for old records.
• Checkpoint reliability risks: The stream’s checkpoint tracks which Delta transactions have been processed. If the checkpoint becomes corrupted or is deleted, the stream will reprocess all historical changes from the Delta table, potentially flooding your ORC storage with duplicate data. Ensure checkpoint storage is secure and backed up.

A quick note on your original setup #

Your original approach reads raw ORC files and branches the stream to both Upsert to Delta and write to another ORC location. Switching to reading the Delta table’s changes is a valid pattern, but make sure you enable CDF if you need to capture updates, and address the potential issues above to maintain end-to-end data integrity.

内容的提问来源于stack exchange，提问作者Tim