{"id":834,"date":"2025-09-07T14:07:06","date_gmt":"2025-09-07T14:07:06","guid":{"rendered":"https:\/\/dataopsschool.com\/blog\/?p=834"},"modified":"2025-09-07T14:07:09","modified_gmt":"2025-09-07T14:07:09","slug":"databricks-hands-on-tutorial-for-dlt-data-quality-expectations","status":"publish","type":"post","link":"https:\/\/dataopsschool.com\/blog\/databricks-hands-on-tutorial-for-dlt-data-quality-expectations\/","title":{"rendered":"Databricks: hands-on tutorial for DLT Data Quality & Expectations"},"content":{"rendered":"\n

Here\u2019s a complete, hands-on tutorial<\/strong> for DLT Data Quality & Expectations<\/strong> \u2014 including how to define rules<\/strong>, use warning \/ fail \/ drop<\/strong> actions, and monitor a DLT pipeline with SQL<\/strong> for observability. I\u2019ve aligned the flow (and examples) to your timestamped outline so it reads like a video transcript you can paste into docs, slides, or a blog.<\/p>\n\n\n\n

\n\n\n\n

Introduction<\/h1>\n\n\n\n

In Delta Live Tables (DLT)<\/strong>, Expectations<\/em> are declarative data-quality rules you apply to each record as it flows through your pipeline. You can validate schemas, ranges, enumerations, nullability, referential integrity, and more. DLT then:<\/p>\n\n\n\n

    \n
  • Counts<\/strong> how many rows passed\/failed.<\/li>\n\n\n\n
  • Decides<\/strong> per rule whether to allow<\/strong>, drop<\/strong>, or fail<\/strong> the pipeline.<\/li>\n\n\n\n
  • Surfaces metrics<\/strong> and event logs<\/strong> you can query and visualize.<\/li>\n<\/ul>\n\n\n\n

    We\u2019ll build a small pipeline with Orders<\/strong> and Customers<\/strong> to demonstrate:<\/p>\n\n\n\n

      \n
    • Defining rules once<\/strong> (as a Python dict) and reusing<\/strong> them.<\/li>\n\n\n\n
    • Applying rules in warning<\/strong>, fail<\/strong>, and drop<\/strong> modes.<\/li>\n\n\n\n
    • Monitoring DLT event logs<\/strong> with SQL to track DQ failures over time.<\/li>\n<\/ul>\n\n\n\n
      \n

      Pre-reqs (quick):<\/p>\n\n\n\n

        \n
      1. Create a DLT pipeline in Databricks; set Product edition to Advanced<\/strong> (Expectations need Advanced).<\/li>\n\n\n\n
      2. Attach to a cluster policy that allows DLT (or use serverless DLT, if enabled).<\/li>\n\n\n\n
      3. Use a Notebook language: Python<\/strong> for the examples below (SQL equivalents included later).<\/li>\n<\/ol>\n<\/blockquote>\n\n\n\n
        \n\n\n\n

        What are Expectations in Databricks DLT?<\/h1>\n\n\n\n

        Expectation<\/strong> = named predicate<\/em> + action<\/em>.<\/p>\n\n\n\n

          \n
        • Name<\/strong>: unique within the dataset (table\/view)<\/li>\n\n\n\n
        • Predicate<\/strong>: a boolean SQL expression (e.g., order_price > 0<\/code>)<\/li>\n\n\n\n
        • Action<\/strong> (per rule):\n
            \n
          • Warning (default)<\/strong>: log failures; do not<\/strong> block or drop rows<\/li>\n\n\n\n
          • Drop<\/strong>: exclude<\/strong> failing rows from the output<\/li>\n\n\n\n
          • Fail<\/strong>: stop<\/strong> the pipeline (task fails)<\/li>\n<\/ul>\n<\/li>\n<\/ul>\n\n\n\n

            Expectations can be attached to:<\/p>\n\n\n\n

              \n
            • DLT Tables<\/strong> (@dlt.table<\/code>)<\/li>\n\n\n\n
            • DLT Views<\/strong> (@dlt.view<\/code>)<\/li>\n\n\n\n
            • Joined\/derived datasets<\/strong> (so you can validate post-join<\/em> quality, too)<\/li>\n<\/ul>\n\n\n\n
              \n\n\n\n

              How to define rules for Expectations in DLT?<\/h1>\n\n\n\n

              Below, we\u2019ll define reusable rule sets as Python dictionaries<\/strong>, then attach them with DLT decorators.<\/p>\n\n\n\n

              Example datasets (conceptual)<\/h2>\n\n\n\n
                \n
              • Orders<\/strong>: order_key<\/code>, order_status<\/code> in ('O','F','P'<\/code>), order_price<\/code> > 0<\/li>\n\n\n\n
              • Customers<\/strong>: cust_id<\/code>, mkt_segment<\/code> NOT NULL<\/strong><\/li>\n<\/ul>\n\n\n\n
                \n

                You can source data from Autoloader, streams, or tables. For clarity, we\u2019ll read from bronze tables (or views) you maintain in your setup notebook.<\/p>\n<\/blockquote>\n\n\n\n

                Python: rule dictionaries<\/h3>\n\n\n\n
                import dlt\nfrom pyspark.sql import functions as F\n\n# Reusable rule sets\norder_rules = {\n  \"valid_order_status\": \"order_status IN ('O','F','P')\",\n  \"valid_order_price\":  \"order_price > 0\"\n}\n\ncustomer_rules = {\n  \"valid_mkt_segment\": \"mkt_segment IS NOT NULL\"\n}\n<\/code><\/pre>\n\n\n\n
                \n
                Tips<\/p>\n\n\n\n
                  \n
                • Keep rule names snake_case<\/strong> and descriptive.<\/li>\n\n\n\n
                • Expressions are SQL strings<\/strong> (not Python boolean expressions).<\/li>\n\n\n\n
                • You can define one or many<\/strong> rules per dataset.<\/li>\n<\/ul>\n<\/blockquote>\n\n\n\n
                  \n\n\n\n
                  Expectation action: Warning<\/h1>\n\n\n\n
                  Warning<\/strong> is the default behavior when you attach expectations with @dlt.expect<\/code> or @dlt.expect_all<\/code>. DLT records failures in metrics<\/strong> and event logs<\/strong>, but keeps<\/strong> the rows.<\/p>\n\n\n\n
                  Python: apply rules in warning<\/strong> mode<\/h3>\n\n\n\n
                  @dlt.table(name=\"orders_bronze\")\n@dlt.expect_all(order_rules)         # default = WARNING (aka \"allow\")\ndef orders_bronze():\n    # Replace this with your real source (e.g., cloud_files or a table)\n    return spark.table(\"source.orders_raw\")\n\n@dlt.view(name=\"customers_v\")\n@dlt.expect_all(customer_rules)      # default = WARNING\ndef customers_v():\n    return spark.table(\"source.customers_raw\")\n<\/code><\/pre>\n\n\n\n
                  What happens at runtime?<\/h3>\n\n\n\n
                    \n
                  • DLT counts<\/strong> failures for each rule and shows them under the dataset\u2019s Data quality<\/strong> tab.<\/li>\n\n\n\n
                  • Rows still land<\/strong> in orders_bronze<\/code> \/ are visible in downstream datasets.<\/li>\n<\/ul>\n\n\n\n
                    \n
                    Use case: Profiling<\/strong> a new source without disrupting downstream consumers. Warning helps you discover<\/strong> issues first.<\/p>\n<\/blockquote>\n\n\n\n
                    \n\n\n\n
                    Expectation action: Fail<\/h1>\n\n\n\n
                    Use fail<\/strong> when any failing row should stop<\/strong> your pipeline. Great for strict contracts<\/strong> and mission-critical<\/strong> pipelines.<\/p>\n\n\n\n
                    Python: fail<\/strong> per rule set<\/h3>\n\n\n\n
                    @dlt.table(name=\"orders_bronze_fail_stop\")\n@dlt.expect_all_or_fail(order_rules)  # if any rule fails => task fails\ndef orders_bronze_fail_stop():\n    return spark.table(\"source.orders_raw\")\n<\/code><\/pre>\n\n\n\n
                    \n
                    Alternative: per-rule fail<\/p>\n<\/blockquote>\n\n\n\n
                    @dlt.table(name=\"orders_bronze_fail_rule\")\n@dlt.expect_or_fail(\"valid_order_status\", \"order_status IN ('O','F','P')\")\n@dlt.expect(\"valid_order_price\", \"order_price > 0\")  # warning for this one\ndef orders_bronze_fail_rule():\n    return spark.table(\"source.orders_raw\")\n<\/code><\/pre>\n\n\n\n
                    \n
                    Use case: SLA-backed<\/strong> pipelines where invalid data must never<\/strong> progress.<\/p>\n<\/blockquote>\n\n\n\n
                    \n\n\n\n
                    Expectation action: Drop<\/h1>\n\n\n\n
                    Use drop<\/strong> when you want to filter out<\/strong> bad rows but keep the pipeline running. This is common in schema drift<\/strong> or edge-case<\/strong> scenarios.<\/p>\n\n\n\n
                    Python: drop<\/strong> per rule set<\/h3>\n\n\n\n
                    @dlt.table(name=\"orders_bronze_dropped\")\n@dlt.expect_all_or_drop(order_rules)   # failing rows are excluded\ndef orders_bronze_dropped():\n    return spark.table(\"source.orders_raw\")\n<\/code><\/pre>\n\n\n\n
                    \n
                    Alternative: per-rule drop<\/p>\n<\/blockquote>\n\n\n\n
                    @dlt.table(name=\"orders_bronze_mixed\")\n@dlt.expect_or_drop(\"valid_order_status\", \"order_status IN ('O','F','P')\")\n@dlt.expect(\"valid_order_price\", \"order_price > 0\")  # warning for price\ndef orders_bronze_mixed():\n    return spark.table(\"source.orders_raw\")\n<\/code><\/pre>\n\n\n\n
                    \n
                    Use case: allow the pipeline to continue<\/strong> while quarantining<\/strong> bad records (e.g., route them later for remediation).<\/p>\n<\/blockquote>\n\n\n\n
                    \n\n\n\n
                    Applying expectations on joins \/ views<\/h2>\n\n\n\n
                    You can attach expectations after transformations<\/strong> (e.g., joins), which validates post-join<\/strong> quality:<\/p>\n\n\n\n
                    @dlt.view(name=\"orders_customers_join_v\")\n@dlt.expect_all({\n  \"post_join_status\": \"order_status IN ('O','F','P')\",\n  \"post_join_price\":  \"order_price > 0\",\n  \"post_join_segment\": \"mkt_segment IS NOT NULL\"\n})\ndef orders_customers_join_v():\n    orders = dlt.read(\"orders_bronze_dropped\")\n    custs  = dlt.read_stream(\"customers_v\")   # view can be read as stream\n    return (orders.alias(\"o\")\n                  .join(custs.alias(\"c\"), F.col(\"o.cust_id\")==F.col(\"c.cust_id\"), \"left\"))\n<\/code><\/pre>\n\n\n\n
                    \n\n\n\n
                    SQL equivalents (DLT SQL)<\/h2>\n\n\n\n
                    If you prefer DLT SQL<\/strong> in the same pipeline:<\/p>\n\n\n\n
                    -- Warning (default)\nCREATE OR REFRESH STREAMING LIVE TABLE orders_bronze_sql\nCONSTRAINT valid_order_status EXPECT (order_status IN ('O','F','P'))\nCONSTRAINT valid_order_price  EXPECT (order_price > 0)\nAS SELECT * FROM source.orders_raw;\n\n-- Fail\nCREATE OR REFRESH STREAMING LIVE TABLE orders_bronze_fail_sql\nCONSTRAINT valid_order_status EXPECT (order_status IN ('O','F','P')) ON VIOLATION FAIL UPDATE\nCONSTRAINT valid_order_price  EXPECT (order_price > 0)               ON VIOLATION FAIL UPDATE\nAS SELECT * FROM source.orders_raw;\n\n-- Drop\nCREATE OR REFRESH STREAMING LIVE TABLE orders_bronze_drop_sql\nCONSTRAINT valid_order_status EXPECT (order_status IN ('O','F','P')) ON VIOLATION DROP ROW\nCONSTRAINT valid_order_price  EXPECT (order_price > 0)               ON VIOLATION DROP ROW\nAS SELECT * FROM source.orders_raw;\n<\/code><\/pre>\n\n\n\n
                    \n
                    Notes:<\/p>\n\n\n\n
                      \n
                    • The SQL syntax uses CONSTRAINT name EXPECT (predicate)<\/code> with ON VIOLATION DROP ROW | FAIL UPDATE<\/code>.<\/li>\n\n\n\n
                    • If you omit ON VIOLATION<\/code>, it behaves like warning<\/strong> (allow).<\/li>\n<\/ul>\n<\/blockquote>\n\n\n\n
                      \n\n\n\n
                      Inserting test records (quick setup idea)<\/h2>\n\n\n\n
                      In a separate setup notebook<\/strong>, you can seed edge cases to see each action:<\/p>\n\n\n\n
                      -- example seed data; in practice, your bronze comes from Autoloader or ingestion jobs\nCREATE TABLE IF NOT EXISTS source.orders_raw AS\nSELECT * FROM VALUES\n  (1001, 'O',  125.00,  1),\n  (1002, 'NA',  30.00,  2),     -- invalid status\n  (1003, 'P',  -10.00,  3),     -- invalid price\n  (1004, NULL,  99.99,  4)      -- invalid status (NULL)\nAS T(order_key, order_status, order_price, cust_id);\n\nCREATE TABLE IF NOT EXISTS source.customers_raw AS\nSELECT * FROM VALUES\n  (1, 'SMALL BIZ'),\n  (2, NULL),                    -- invalid segment\n  (3, 'ENTERPRISE'),\n  (4, 'CONSUMER')\nAS T(cust_id, mkt_segment);\n<\/code><\/pre>\n\n\n\n
                      Re-run the pipeline to observe warning\/fail\/drop<\/strong> effects.<\/p>\n\n\n\n
                      \n\n\n\n
                      How to Monitor DLT pipelines (Observability)<\/h1>\n\n\n\n
                      DLT emits rich event logs<\/strong> you can query with SQL<\/strong> to build dashboards<\/strong> (SQL Warehouse, Databricks Dashboards, Lakeview, etc.).<\/p>\n\n\n\n
                      \n
                      Find the Pipeline ID<\/strong>: open your DLT pipeline \u2192 copy the ID from the UI.<\/p>\n<\/blockquote>\n\n\n\n
                      1) Raw event log (table-valued function)<\/h3>\n\n\n\n
                      -- Replace with your pipeline id\nSELECT * FROM event_log('<PIPELINE_ID>');\n<\/code><\/pre>\n\n\n\n
                      2) Create helpful views for reuse<\/h3>\n\n\n\n
                      -- Raw view (persisted in your schema)\nCREATE OR REPLACE VIEW dlt_event_log_raw AS\nSELECT * FROM event_log('<PIPELINE_ID>');\n\n-- Latest update per dataset\nCREATE OR REPLACE TEMP VIEW dlt_latest_updates AS\nSELECT\n  id:flow_definition.output_dataset AS dataset_name,\n  MAX(timestamp)                     AS last_update_ts\nFROM dlt_event_log_raw\nWHERE event_type = 'flow_progress'\nGROUP BY 1;\n<\/code><\/pre>\n\n\n\n
                      3) Data Quality (DQ) metrics over time<\/h3>\n\n\n\n
                      DLT emits a nested expectations<\/code> structure inside events (commonly in flow_progress<\/code> and expectations_status<\/code>). This query summarizes failures<\/strong> by dataset and rule name:<\/p>\n\n\n\n
                      WITH dq AS (\n  SELECT\n    timestamp,\n    dataset = COALESCE(\n      id:flow_definition.output_dataset::string,\n      details:flow_progress:output_dataset::string\n    ),\n    explode(expectations) AS exp\n  FROM dlt_event_log_raw\n  WHERE event_type IN ('flow_progress','expectations_status')\n),\nexp_flat AS (\n  SELECT\n    timestamp,\n    dataset,\n    exp.key                                  AS rule_name,\n    exp.value.passed_records::long           AS passed_records,\n    exp.value.failed_records::long           AS failed_records,\n    exp.value.status::string                 AS action -- ALLOW \/ DROP \/ FAIL\n  FROM dq\n)\nSELECT\n  dataset,\n  rule_name,\n  action,\n  SUM(passed_records) AS total_passed,\n  SUM(failed_records) AS total_failed,\n  MAX(timestamp)      AS last_seen\nFROM exp_flat\nGROUP BY dataset, rule_name, action\nORDER BY dataset, rule_name;\n<\/code><\/pre>\n\n\n\n
                      \n
                      What you\u2019ll see:<\/p>\n\n\n\n
                        \n
                      • action<\/code><\/strong> per rule (ALLOW<\/code> ~ warning, DROP<\/code>, FAIL<\/code>)<\/li>\n\n\n\n
                      • passed\/failed<\/strong> counts and last_seen<\/strong> timestamp<\/li>\n\n\n\n
                      • Use this to power SLO dashboards<\/strong> (e.g., \u201c<1% failure over last 7 days\u201d)<\/li>\n<\/ul>\n<\/blockquote>\n\n\n\n
                        4) Pull the latest failure samples (optional)<\/h3>\n\n\n\n
                        To peek at failing rows<\/strong>, design your bronze with a quarantine table<\/strong> or DLT expectations with DROP<\/strong> + audit<\/strong>:<\/p>\n\n\n\n
                          \n
                        • If using DROP<\/strong>, write failing rows to a side sink (custom logic in bronze).<\/li>\n\n\n\n
                        • Or, add trace columns<\/strong> (ingest time, source file, etc.) to speed triage.<\/li>\n<\/ul>\n\n\n\n
                          5) Operational health<\/h3>\n\n\n\n
                          Summarize task status<\/strong>, update durations<\/strong>, and throughput<\/strong>:<\/p>\n\n\n\n
                          -- Task runs and statuses\nSELECT\n  timestamp,\n  details:flow_progress:metrics:num_output_rows::long AS out_rows,\n  details:flow_progress:metrics:input_rows_processed::long AS in_rows,\n  details:flow_progress:status::string AS status,\n  id:flow_definition.output_dataset::string AS dataset\nFROM dlt_event_log_raw\nWHERE event_type = 'flow_progress'\nORDER BY timestamp DESC;\n<\/code><\/pre>\n\n\n\n
                          \n
                          Build visuals:<\/p>\n\n\n\n
                            \n
                          • Bar<\/strong>: failed_records by rule_name over time<\/li>\n\n\n\n
                          • Line<\/strong>: throughput (rows\/sec) vs update<\/li>\n\n\n\n
                          • KPI<\/strong>: last update latency \/ time since last success<\/li>\n<\/ul>\n<\/blockquote>\n\n\n\n
                            \n\n\n\n
                            Practical guidance & patterns<\/h2>\n\n\n\n
                              \n
                            • Start in Warning<\/strong> \u2192 learn the data \u2192 tighten<\/strong> rules over time \u2192 promote to Drop<\/strong> or Fail<\/strong> where appropriate.<\/li>\n\n\n\n
                            • Name rules<\/strong> predictably: valid_<col>_<constraint><\/code> (valid_order_price_gt_zero<\/code>).<\/li>\n\n\n\n
                            • Centralize rules<\/strong> (dicts or SQL macros) to reuse<\/strong> across datasets.<\/li>\n\n\n\n
                            • Post-join validations<\/strong> catch issues that only appear after enrichment.<\/li>\n\n\n\n
                            • Quarantine patterns<\/strong>: When using DROP<\/strong>, optionally write failures<\/strong> to a side table for remediation (e.g., with a stream that matches the inverse predicate).<\/li>\n\n\n\n
                            • Alerting<\/strong>: Schedule SQL alerts<\/strong> (e.g., failed_records > 0) or use Jobs<\/strong> to run health checks and send notifications (email\/Webhook\/Slack).<\/li>\n<\/ul>\n\n\n\n
                              \n\n\n\n
                              Troubleshooting<\/h2>\n\n\n\n
                                \n
                              • Wrong column name<\/strong> in a rule? DLT will fail early<\/strong> and show it in Event log<\/strong>. Fix the expression and re-run.<\/li>\n\n\n\n
                              • Nothing shows up<\/strong> in metrics? Ensure the dataset with expectations actually runs<\/strong> (new data or full refresh), and you\u2019re viewing the right pipeline\u2019s<\/strong> event log \/ catalog.<\/li>\n<\/ul>\n\n\n\n
                                \n\n\n\n
                                One-page checklist<\/h2>\n\n\n\n
                                  \n
                                • Edition set to Advanced<\/strong><\/li>\n\n\n\n
                                • Rule dicts defined and unit-tested<\/strong> on small slices<\/li>\n\n\n\n
                                • Expectations applied at bronze<\/strong>, post-join<\/strong>, silver<\/strong> as needed<\/li>\n\n\n\n
                                • Warning<\/strong> for discovery, Drop<\/strong> to quarantine, Fail<\/strong> for contracts<\/li>\n\n\n\n
                                • Event log<\/strong> views created; DQ summary<\/strong> query saved<\/li>\n\n\n\n
                                • Dashboards<\/strong> and alerts<\/strong> wired to your SLOs<\/li>\n<\/ul>\n\n\n\n
                                  \n\n\n\n
                                  <\/p>\n","protected":false},"excerpt":{"rendered":"
                                  Here\u2019s a complete, hands-on tutorial for DLT Data Quality & Expectations \u2014 including how to define rules, use warning \/ fail \/ drop actions, and monitor 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-834","post","type-post","status-publish","format-standard","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/posts\/834","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=834"}],"version-history":[{"count":1,"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/posts\/834\/revisions"}],"predecessor-version":[{"id":835,"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/posts\/834\/revisions\/835"}],"wp:attachment":[{"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/media?parent=834"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/categories?post=834"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/dataopsschool.com\/blog\/wp-json\/wp\/v2\/tags?post=834"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}