<\/p>\n\n\n\n
Today we\u2019ll cover four production patterns for Delta Live Tables (DLT)<\/strong>:<\/p>\n\n\n\n Problem:<\/strong> Your upstream system truncates a Delta table and then inserts new data. A \u201cpure\u201d streaming read sees this as non-append changes and will fail.<\/p>\n\n\n\n Fix:<\/strong> When reading a Delta table as a streaming source<\/strong>, set:<\/p>\n\n\n\n Notes<\/strong><\/p>\n\n\n\n Full refresh<\/strong> re-computes & re-loads all<\/strong> pipeline datasets from scratch; streaming tables may be truncated<\/strong> (history lost) before re-build. You can prevent<\/strong> full refresh on selected tables with a table property:<\/p>\n\n\n\n When to use:<\/strong> Protect SCD2<\/strong> or any long-history tables from accidental resets, while still allowing normal incremental updates to flow. (Databricks Documentation<\/a>)<\/p>\n\n\n\n You can orchestrate DLT with Workflows<\/strong> (Jobs). Two common triggers:<\/p>\n\n\n\n Why volumes\/external locations? File-arrival triggers explicitly support Unity Catalog volumes\/external locations<\/strong> for secure, first-class monitoring. (Databricks Documentation<\/a>)<\/p>\n<\/blockquote>\n\n\n\n Switch your DLT pipeline to Production<\/strong> so the job auto-terminates<\/strong> compute after each successful\/failed run (lower cost). This is the recommended mode for scheduled or file-arrival-triggered pipelines.<\/p>\n\n\n\n Tip: The event log<\/strong> contains throughput, rule failures, and status\u2014great for DLT health<\/strong> and SLA reporting. (community.databricks.com<\/a>)<\/p>\n<\/blockquote>\n\n\n\n Then wire this pipeline into a Workflow<\/strong> (Job) with either a Schedule<\/strong> or File-arrival trigger<\/strong> pointing at your landing location.<\/p>\n\n\n\n <\/p>\n","protected":false},"excerpt":{"rendered":" Introduction Today we\u2019ll cover four production patterns for Delta Live Tables (DLT): Truncate-Load table as Source for Streaming Tables (with skipChangeCommits) Problem: Your upstream system truncates a… <\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-836","post","type-post","status-publish","format-standard","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/posts\/836","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/comments?post=836"}],"version-history":[{"count":1,"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/posts\/836\/revisions"}],"predecessor-version":[{"id":837,"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/posts\/836\/revisions\/837"}],"wp:attachment":[{"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/media?parent=836"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/categories?post=836"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/tags?post=836"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}\n
\n\n\n\nTruncate-Load table as Source for Streaming Tables (with
skipChangeCommits<\/code>)<\/h1>\n\n\n\n\n
skipChangeCommits = true<\/code> \u2192 ignore change commits (updates\/deletes\/truncates) and continue from the latest snapshot. (Databricks Documentation<\/a>, Microsoft Learn<\/a>)<\/li>\n<\/ul>\n\n\n\nDLT (Python) example<\/h3>\n\n\n\n
import dlt\nfrom pyspark.sql import functions as F\n\n# Truncate-load source path or table\nSOURCE_TABLE = \"source.orders_raw\" # could also be a UC path like catalog.schema.table\n\n@dlt.table(\n name=\"orders_bronze\",\n comment=\"Bronze from truncate-load source using skipChangeCommits.\"\n)\ndef orders_bronze():\n # If you read by table name:\n return (spark.readStream\n .format(\"delta\")\n .option(\"skipChangeCommits\", \"true\") # KEY!\n .table(SOURCE_TABLE))\n # Or if you read by path:\n # return (spark.readStream.format(\"delta\")\n # .option(\"skipChangeCommits\",\"true\")\n # .load(\"dbfs:\/Volumes\/dev\/etl\/landing\/files\/orders_raw\"))\n<\/code><\/pre>\n\n\n\n\n
readChangeFeed<\/code>) instead. (Databricks Documentation<\/a>)<\/li>\n<\/ul>\n\n\n\n
\n\n\n\nFull Refresh (what it does) & how to avoid<\/strong> refreshing specific tables<\/h1>\n\n\n\n
\n
pipelines.reset.allowed = false<\/code> (default is true<\/code>). (Databricks Documentation<\/a>, Microsoft Learn<\/a>)<\/li>\n<\/ul>\n\n\n\nDLT (Python) decorator with table properties<\/h3>\n\n\n\n
@dlt.table(\n name=\"orders_silver\",\n table_properties={ \"pipelines.reset.allowed\": \"false\" }, # protect from full refresh\n comment=\"Silver table protected from full refresh.\"\n)\ndef orders_silver():\n df = dlt.read(\"orders_bronze\")\n return df.select(\"order_key\",\"order_status\",\"order_price\",\"cust_id\")\n<\/code><\/pre>\n\n\n\nDLT (SQL) equivalent<\/h3>\n\n\n\n
CREATE OR REFRESH STREAMING LIVE TABLE orders_silver_sql\nTBLPROPERTIES (pipelines.reset.allowed = false)\nAS\nSELECT order_key, order_status, order_price, cust_id\nFROM LIVE.orders_bronze;\n<\/code><\/pre>\n\n\n\n
\n\n\n\nScheduling DLT pipelines with Workflows<\/strong><\/h1>\n\n\n\n
\n
A. Add a file-arrival trigger<\/strong> (UI steps)<\/h3>\n\n\n\n
\n
uc:\/\/catalog.schema.volume\/path\/<\/code> or abfss:\/\/\u2026<\/code>).<\/li>\n\n\n\n\n
B. Add a time-based schedule<\/strong><\/h3>\n\n\n\n
\n
C. Run in Production<\/strong> mode<\/h3>\n\n\n\n
\n\n\n\nEnd-to-end example: a one-task Workflow that runs your DLT pipeline<\/h2>\n\n\n\n
\n
orders_bronze<\/code>\/orders_silver<\/code>).<\/li>\n\n\n\n
\n\n\n\nHelpful SQL: identify full-refresh runs & basic observability<\/h2>\n\n\n\n
\n
event_log()<\/code><\/strong> table-valued function. You can create views and dashboards on top of it. (Databricks Documentation<\/a>, Microsoft Learn<\/a>)<\/li>\n<\/ul>\n\n\n\n-- All events\nSELECT * FROM event_log('<PIPELINE_ID>');\n\n-- Was the last run a full refresh?\nSELECT\n MAX(CASE WHEN details:full_refresh::boolean THEN 1 ELSE 0 END) AS was_full_refresh,\n MAX(timestamp) AS last_event_time\nFROM event_log('<PIPELINE_ID>');\n<\/code><\/pre>\n\n\n\n\n
\n\n\n\nTested patterns & gotchas<\/h2>\n\n\n\n
\n
.option(\"skipChangeCommits\",\"true\")<\/code> on the streaming<\/strong> read of a Delta source that is not strictly append-only. This allows you to keep streaming even when the source truncates. (Databricks Documentation<\/a>, Microsoft Learn<\/a>)<\/li>\n\n\n\npipelines.reset.allowed=false<\/code> at the table<\/strong> to prevent accidental full refresh. (Databricks Documentation<\/a>)<\/li>\n\n\n\nevent_log('<PIPELINE_ID>')<\/code> and alert if failures spike or if no successful run happens within your SLO window. (Databricks Documentation<\/a>)<\/li>\n<\/ul>\n\n\n\n
\n\n\n\nQuick copy-paste: minimal pipeline (Python)<\/h2>\n\n\n\n
import dlt\nfrom pyspark.sql import functions as F\n\n# ----- Bronze from truncate-load source -----\n@dlt.table(\n name=\"orders_bronze\",\n comment=\"Bronze from truncate-load source\"\n)\ndef orders_bronze():\n return (spark.readStream\n .format(\"delta\")\n .option(\"skipChangeCommits\", \"true\") # key for truncate-load\n .table(\"source.orders_raw\"))\n\n# ----- Silver protected from full refresh -----\n@dlt.table(\n name=\"orders_silver\",\n table_properties={\"pipelines.reset.allowed\": \"false\"},\n comment=\"Silver table protected from full refresh\"\n)\ndef orders_silver():\n df = dlt.read(\"orders_bronze\")\n return (df\n .withColumn(\"order_price_usd\", F.col(\"order_price\").cast(\"double\"))\n .filter(F.col(\"order_price_usd\") > 0))\n<\/code><\/pre>\n\n\n\n
\n\n\n\nReferences<\/h3>\n\n\n\n
\n
pipelines.reset.allowed<\/code>, auto-optimize, etc.). (Databricks Documentation<\/a>, Microsoft Learn<\/a>)<\/li>\n\n\n\nskipChangeCommits<\/code> option for streaming reads. (Databricks Documentation<\/a>, Microsoft Learn<\/a>)<\/li>\n\n\n\n
\n\n\n\n