From 86f79c3afe81839b80cda1fc69d97135f5105b04 Mon Sep 17 00:00:00 2001 From: dinbab1984 Date: Wed, 6 Aug 2025 22:20:56 +0200 Subject: [PATCH 01/16] Issue#489 - Feature - DQX Spark Structured Streaming Demo --- demos/dqx_streaming_demo.py | 227 ++++++++++++++++++++++++++++++++++++ docs/dqx/docs/demos.mdx | 1 + 2 files changed, 228 insertions(+) create mode 100644 demos/dqx_streaming_demo.py diff --git a/demos/dqx_streaming_demo.py b/demos/dqx_streaming_demo.py new file mode 100644 index 000000000..91b63828b --- /dev/null +++ b/demos/dqx_streaming_demo.py @@ -0,0 +1,227 @@ +# Databricks notebook source +# MAGIC +# MAGIC %md +# MAGIC ### Environment Setup for DQX with Structured Streaming +# MAGIC +# MAGIC #### 1\. Library Installation +# MAGIC +# MAGIC **Option A — Notebook-scoped installation** +# MAGIC +# MAGIC - Use notebook-scoped pip installs, which take effect only for the current interactive notebook session. +# MAGIC +# MAGIC ```py +# MAGIC # Install DQX library in the current notebook session +# MAGIC # Run this cell before any imports from databricks-labs-dqx +# MAGIC +# MAGIC %pip install databricks-labs-dqx +# MAGIC ``` +# MAGIC +# MAGIC - If you are using a cluster that was running prior to this install, you may need to restart the Python process for the changes to take full effect: +# MAGIC +# MAGIC ```py +# MAGIC # Optional: Force restart of the Python process if required by your environment +# MAGIC dbutils.library.restartPython() +# MAGIC ``` +# MAGIC +# MAGIC **Option B — Cluster-scoped installation (recommended for production streaming jobs)** +# MAGIC +# MAGIC - Install the DQX library to your cluster via the Databricks Libraries UI or by specifying it as a PyPI library dependency during cluster creation. +# MAGIC - This ensures the package is available for all jobs or notebooks attached to the cluster, including those running streaming workloads. +# MAGIC - The syntax for specifying the package is: +# MAGIC - Library Source: PyPI +# MAGIC - Package: `databricks-labs-dqx` +# MAGIC + +# COMMAND ---------- + +from databricks.labs.dqx.engine import DQEngine +from databricks.sdk import WorkspaceClient +from pyspark.sql import DataFrame +from pyspark.sql import DataFrame +from pyspark.sql.streaming import DataStreamWriter + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ### Bronze Stream +# MAGIC +# MAGIC This cell reads the NYC Taxi 2019 dataset from a Delta table as a streaming DataFrame, which will be used as the input for downstream data quality checks and transformations. + +# COMMAND ---------- + +bronze_stream = ( + spark.readStream + .format("delta") + .load("/databricks-datasets/delta-sharing/samples/nyctaxi_2019") +) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Define Data Quality Checks +# MAGIC +# MAGIC This cell defines a set of data quality checks using YAML syntax. The checks are designed to validate key columns in the NYC Taxi dataset, such as `vendor_id`, `pickup_datetime`, `dropoff_datetime`, `passenger_count`, and `trip_distance`. Each check specifies a function, its arguments, a name, and a criticality level (error or warn). These checks will be used by the DQEngine to enforce data quality rules on the streaming data. + +# COMMAND ---------- + +# Define Data Quality checks +import yaml + + +checks = yaml.safe_load(""" +- check: + function: is_not_null + arguments: + column: vendor_id + name: vendor_id_is_null + criticality: error +- check: + function: is_not_null_and_not_empty + arguments: + column: vendor_id + trim_strings: true + name: vendor_id_is_null_or_empty + criticality: error + +- check: + function: is_not_null + arguments: + column: pickup_datetime + name: pickup_datetime_is_null + criticality: error +- check: + function: is_not_in_future + arguments: + column: pickup_datetime + name: pickup_datetime_isnt_in_range + criticality: warn + +- check: + function: is_not_in_future + arguments: + column: pickup_datetime + name: pickup_datetime_not_in_future + criticality: warn +- check: + function: is_not_in_future + arguments: + column: dropoff_datetime + name: dropoff_datetime_not_in_future + criticality: warn +- check: + function: is_not_null + arguments: + column: passenger_count + name: passenger_count_is_null + criticality: error +- check: + function: is_in_range + arguments: + column: passenger_count + min_limit: 0 + max_limit: 6 + name: passenger_incorrect_count + criticality: warn +- check: + function: is_not_null + arguments: + column: trip_distance + name: trip_distance_is_null + criticality: error +""") + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Apply data quality checks to the bronze_stream DataFrame using the defined checks metadata +# MAGIC The resulting DataFrame, bronze_dq_checked, contains data with quality check results attached + +# COMMAND ---------- + +dq_engine = DQEngine(WorkspaceClient()) + +def apply_data_quality(df: DataFrame) -> DataFrame: + return dq_engine.apply_checks_by_metadata(df, checks) + +bronze_dq_checked = apply_data_quality(bronze_stream) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Extract Valid Records for Silver Table +# MAGIC +# MAGIC This cell defines a function `get_valid` that extracts valid records from the data quality-checked streaming DataFrame using the DQEngine. The resulting DataFrame, `silver_df`, contains only records that passed all data quality checks and is ready for further processing or writing to the silver table. + +# COMMAND ---------- + +def get_valid(df: DataFrame) -> DataFrame: + return dq_engine.get_valid(df) + +silver_df = get_valid(bronze_dq_checked) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Extract Invalid Records for Quarantine Table +# MAGIC +# MAGIC This cell defines a function `get_invalid` that extracts invalid records from the data quality-checked streaming DataFrame using the DQEngine. The resulting DataFrame, `quarantine_df`, contains only records that did not pass all data quality checks and is not ready for further processing or writing to the quarantine table. + +# COMMAND ---------- + +def get_invalid(df: DataFrame) -> DataFrame: + return dq_engine.get_invalid(df) + +quarantine_df = get_invalid(bronze_dq_checked) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Write streaming DataFrames to Delta tables with checkpointing +# MAGIC - silver_df: Valid records after data quality checks +# MAGIC - quarantine_df: Invalid records (quarantined) +# MAGIC +# MAGIC and +# MAGIC - Uses availableNow trigger for batch-like streaming +# MAGIC - Checkpoint and table locations are set via widgets + +# COMMAND ---------- + +dbutils.widgets.text("silver_checkpoint", "/tmp/dq/structured/bronze_to_silver_checkpoint") +dbutils.widgets.text("silver_table", "/tmp/dq/structured/silver_table") +dbutils.widgets.text("quarantine_checkpoint", "/tmp/dq/structured/bronze_to_quarantine_checkpoint") +dbutils.widgets.text("quarantine_table", "/tmp/dq/structured/quarantine_table") + +# COMMAND ---------- + +silver_checkpoint = dbutils.widgets.get("silver_checkpoint") +silver_table = dbutils.widgets.get("silver_table") +quarantine_checkpoint = dbutils.widgets.get("quarantine_checkpoint") +quarantine_table = dbutils.widgets.get("quarantine_table") + +# COMMAND ---------- + +# Function to write a streaming DataFrame to a Delta table with checkpointing +def write_stream(df, checkpoint_location, target): + # Configure the streaming write with Delta format, append mode, checkpointing, and trigger + writer = ( + df.writeStream + .format("delta") + .outputMode("append") + .option("checkpointLocation", checkpoint_location) + .trigger(availableNow=True) + ) + # Use toTable for managed tables, start for path-based tables + if target.startswith("/") or target.startswith("dbfs:/"): + return writer.start(target) + else: + return writer.toTable(target) + +# Start streaming queries for silver and quarantine tables +silver_query: DataStreamWriter = write_stream(silver_df, silver_checkpoint, silver_table) +quarantine_query: DataStreamWriter = write_stream(quarantine_df, quarantine_checkpoint, quarantine_table) + +# COMMAND ---------- + +# Wait for the streams to finish or keep them running for interactive sessions if required +silver_query.awaitTermination() +quarantine_query.awaitTermination() \ No newline at end of file diff --git a/docs/dqx/docs/demos.mdx b/docs/dqx/docs/demos.mdx index 7c4f7b35c..fcfd82824 100644 --- a/docs/dqx/docs/demos.mdx +++ b/docs/dqx/docs/demos.mdx @@ -11,6 +11,7 @@ Import the following notebooks in the Databricks workspace to try DQX out: * [DQX Quick Start Demo Notebook (library)](https://github.com/databrickslabs/dqx/blob/v0.7.1/demos/dqx_quick_start_demo_library.py) - quickstart on how to use DQX as a library. * [DQX Demo Notebook (library)](https://github.com/databrickslabs/dqx/blob/v0.7.1/demos/dqx_demo_library.py) - demonstrates how to use DQX as a library. * [DQX Lakeflow Pipelines Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.7.1/demos/dqx_dlt_demo.py) - demonstrates how to use DQX with Lakeflow Pipelines (formerly Delta Live Tables (DLT)). +* [DQX Spark Structured Streaming Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.7.1/demos/dqx_streaming_demo.py) - demonstrates how to use DQX with Spark Structured Streaming. * [DQX Asset Bundles Demo](https://github.com/databrickslabs/dqx/blob/v0.7.1/demos/dqx_demo_asset_bundle/README.md) - demonstrates how to use DQX as a library with Databricks Asset Bundles. * [DQX Demo with dbt](https://github.com/databrickslabs/dqx/blob/v0.7.1/demos/dqx_demo_dbt/README.md) - demonstrates how to use DQX as a library with dbt projects. From cb21e4699f4a2c0fbf9085fdade7f98c057b6f58 Mon Sep 17 00:00:00 2001 From: dinbab1984 Date: Thu, 7 Aug 2025 20:07:34 +0200 Subject: [PATCH 02/16] Issue#489 - Feature - DQX Spark Structured Streaming Demo --- demos/dqx_streaming_demo.py | 17 +++++++++++++++++ tests/e2e/test_run_demos.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/demos/dqx_streaming_demo.py b/demos/dqx_streaming_demo.py index 91b63828b..50726d4cb 100644 --- a/demos/dqx_streaming_demo.py +++ b/demos/dqx_streaming_demo.py @@ -34,6 +34,23 @@ # COMMAND ---------- +# MAGIC %md +# MAGIC ### Install DQX as Library
+# MAGIC For this demo, we will install DQX as library + +# COMMAND ---------- + +dbutils.widgets.text("test_library_ref", "", "Test Library Ref") + +if dbutils.widgets.get("test_library_ref") != "": + %pip install '{dbutils.widgets.get("test_library_ref")}' +else: + %pip install databricks-labs-dqx + +%restart_python + +# COMMAND ---------- + from databricks.labs.dqx.engine import DQEngine from databricks.sdk import WorkspaceClient from pyspark.sql import DataFrame diff --git a/tests/e2e/test_run_demos.py b/tests/e2e/test_run_demos.py index f9d87ef15..1ded8fae1 100644 --- a/tests/e2e/test_run_demos.py +++ b/tests/e2e/test_run_demos.py @@ -3,6 +3,7 @@ from datetime import timedelta from pathlib import Path +from uuid import uuid4 from databricks.sdk import WorkspaceClient from databricks.sdk.service.workspace import ImportFormat from databricks.sdk.service.pipelines import NotebookLibrary, PipelinesEnvironment, PipelineLibrary @@ -191,3 +192,32 @@ def validate_demo_run_status(run: Run, client: WorkspaceClient) -> None: assert ( termination_details.type == TerminationTypeType.SUCCESS ), f"Run of '{task.task_key}' failed with message: {client.jobs.get_run_output(task.run_id).error}" + + +def test_run_dqx_streaming_demo(make_notebook, make_job, tmp_path): + + ws = WorkspaceClient() + path = Path(__file__).parent.parent.parent / "demos" / "dqx_streaming_demo.py" + with open(path, "rb") as f: + notebook = make_notebook(content=f, format=ImportFormat.SOURCE) + notebook_path = notebook.as_fuse().as_posix() + + # Use the temporary directory for outputs + run_id = str(uuid4()) + base_output_path = tmp_path / run_id + base_parameters = { + "silver_checkpoint": f"{base_output_path}/silver_checkpoint", + "silver_table": f"{base_output_path}/silver_table", + "quarantine_checkpoint": f"{base_output_path}/quarantine_checkpoint", + "quarantine_table": f"{base_output_path}/quarantine_table", + "test_library_ref": TEST_LIBRARY_REF, + } + notebook_task = NotebookTask(notebook_path=notebook_path, base_parameters=base_parameters) + job = make_job(tasks=[Task(task_key="dqx_streaming_demo", notebook_task=notebook_task)]) + waiter = ws.jobs.run_now_and_wait(job.job_id) + run = ws.jobs.wait_get_run_job_terminated_or_skipped( + run_id=waiter.run_id, + timeout=timedelta(minutes=30), + callback=lambda r: validate_demo_run_status(r, client=ws), + ) + logging.info(f"Job run {run.run_id} completed successfully for dqx_streaming_demo") From 2ca7150f68ceea5436a25e21d365c3308d933d74 Mon Sep 17 00:00:00 2001 From: dinbab1984 Date: Thu, 7 Aug 2025 22:29:12 +0200 Subject: [PATCH 03/16] Issue#489 - Feature - DQX Spark Structured Streaming Demo --- demos/dqx_streaming_demo.py | 92 +++++++++++++------------------------ 1 file changed, 31 insertions(+), 61 deletions(-) diff --git a/demos/dqx_streaming_demo.py b/demos/dqx_streaming_demo.py index 50726d4cb..026bd2589 100644 --- a/demos/dqx_streaming_demo.py +++ b/demos/dqx_streaming_demo.py @@ -37,6 +37,7 @@ # MAGIC %md # MAGIC ### Install DQX as Library
# MAGIC For this demo, we will install DQX as library +# MAGIC # COMMAND ---------- @@ -51,14 +52,6 @@ # COMMAND ---------- -from databricks.labs.dqx.engine import DQEngine -from databricks.sdk import WorkspaceClient -from pyspark.sql import DataFrame -from pyspark.sql import DataFrame -from pyspark.sql.streaming import DataStreamWriter - -# COMMAND ---------- - # MAGIC %md # MAGIC ### Bronze Stream # MAGIC @@ -75,7 +68,7 @@ # COMMAND ---------- # MAGIC %md -# MAGIC ## Define Data Quality Checks +# MAGIC ### Define Data Quality Checks # MAGIC # MAGIC This cell defines a set of data quality checks using YAML syntax. The checks are designed to validate key columns in the NYC Taxi dataset, such as `vendor_id`, `pickup_datetime`, `dropoff_datetime`, `passenger_count`, and `trip_distance`. Each check specifies a function, its arguments, a name, and a criticality level (error or warn). These checks will be used by the DQEngine to enforce data quality rules on the streaming data. @@ -150,76 +143,51 @@ # COMMAND ---------- # MAGIC %md -# MAGIC ## Apply data quality checks to the bronze_stream DataFrame using the defined checks metadata -# MAGIC The resulting DataFrame, bronze_dq_checked, contains data with quality check results attached - -# COMMAND ---------- - -dq_engine = DQEngine(WorkspaceClient()) - -def apply_data_quality(df: DataFrame) -> DataFrame: - return dq_engine.apply_checks_by_metadata(df, checks) - -bronze_dq_checked = apply_data_quality(bronze_stream) +# MAGIC ### Inputs +# MAGIC - Checkpoint and table locations are set via widgets # COMMAND ---------- -# MAGIC %md -# MAGIC ## Extract Valid Records for Silver Table -# MAGIC -# MAGIC This cell defines a function `get_valid` that extracts valid records from the data quality-checked streaming DataFrame using the DQEngine. The resulting DataFrame, `silver_df`, contains only records that passed all data quality checks and is ready for further processing or writing to the silver table. +dbutils.widgets.text("silver_checkpoint", "/tmp/dq/structured/bronze_to_silver_checkpoint") +dbutils.widgets.text("quarantine_checkpoint", "/tmp/dq/structured/bronze_to_quarantine_checkpoint") +dbutils.widgets.text("silver_table", "/tmp/dq/structured/silver_table") +dbutils.widgets.text("quarantine_table", "/tmp/dq/structured/quarantine_table") # COMMAND ---------- -def get_valid(df: DataFrame) -> DataFrame: - return dq_engine.get_valid(df) - -silver_df = get_valid(bronze_dq_checked) +silver_checkpoint = dbutils.widgets.get("silver_checkpoint") +quarantine_checkpoint = dbutils.widgets.get("quarantine_checkpoint") +silver_table = dbutils.widgets.get("silver_table") +quarantine_table = dbutils.widgets.get("quarantine_table") # COMMAND ---------- # MAGIC %md -# MAGIC ## Extract Invalid Records for Quarantine Table -# MAGIC -# MAGIC This cell defines a function `get_invalid` that extracts invalid records from the data quality-checked streaming DataFrame using the DQEngine. The resulting DataFrame, `quarantine_df`, contains only records that did not pass all data quality checks and is not ready for further processing or writing to the quarantine table. +# MAGIC ### Apply data quality checks and split the bronze stream into silver and quarantine DataFrames # COMMAND ---------- -def get_invalid(df: DataFrame) -> DataFrame: - return dq_engine.get_invalid(df) +from databricks.labs.dqx.engine import DQEngine +from databricks.sdk import WorkspaceClient +from pyspark.sql import DataFrame -quarantine_df = get_invalid(bronze_dq_checked) +dq_engine = DQEngine(WorkspaceClient()) -# COMMAND ---------- +def apply_data_quality_and_split(df: DataFrame): + return dq_engine.apply_checks_by_metadata_and_split(df, checks) -# MAGIC %md -# MAGIC ## Write streaming DataFrames to Delta tables with checkpointing -# MAGIC - silver_df: Valid records after data quality checks -# MAGIC - quarantine_df: Invalid records (quarantined) -# MAGIC -# MAGIC and -# MAGIC - Uses availableNow trigger for batch-like streaming -# MAGIC - Checkpoint and table locations are set via widgets +silver_df, quarantine_df = apply_data_quality_and_split(bronze_stream) # COMMAND ---------- -dbutils.widgets.text("silver_checkpoint", "/tmp/dq/structured/bronze_to_silver_checkpoint") -dbutils.widgets.text("silver_table", "/tmp/dq/structured/silver_table") -dbutils.widgets.text("quarantine_checkpoint", "/tmp/dq/structured/bronze_to_quarantine_checkpoint") -dbutils.widgets.text("quarantine_table", "/tmp/dq/structured/quarantine_table") +# MAGIC %md +# MAGIC ### Write streaming DataFrames to Delta tables with checkpointing # COMMAND ---------- -silver_checkpoint = dbutils.widgets.get("silver_checkpoint") -silver_table = dbutils.widgets.get("silver_table") -quarantine_checkpoint = dbutils.widgets.get("quarantine_checkpoint") -quarantine_table = dbutils.widgets.get("quarantine_table") - -# COMMAND ---------- +PATH_PREFIXES = ("/", "dbfs:/", "s3://", "abfss://", "gs://") -# Function to write a streaming DataFrame to a Delta table with checkpointing def write_stream(df, checkpoint_location, target): - # Configure the streaming write with Delta format, append mode, checkpointing, and trigger writer = ( df.writeStream .format("delta") @@ -227,18 +195,20 @@ def write_stream(df, checkpoint_location, target): .option("checkpointLocation", checkpoint_location) .trigger(availableNow=True) ) - # Use toTable for managed tables, start for path-based tables - if target.startswith("/") or target.startswith("dbfs:/"): + if target.startswith(PATH_PREFIXES): return writer.start(target) else: return writer.toTable(target) -# Start streaming queries for silver and quarantine tables -silver_query: DataStreamWriter = write_stream(silver_df, silver_checkpoint, silver_table) -quarantine_query: DataStreamWriter = write_stream(quarantine_df, quarantine_checkpoint, quarantine_table) +silver_query = write_stream(silver_df, silver_checkpoint, silver_table) +quarantine_query = write_stream(quarantine_df, quarantine_checkpoint, quarantine_table) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ### Wait for the streams to finish or keep them running for interactive sessions if required # COMMAND ---------- -# Wait for the streams to finish or keep them running for interactive sessions if required silver_query.awaitTermination() quarantine_query.awaitTermination() \ No newline at end of file From bfce8648394c8f7ede638542ca0f37366d9005d8 Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Fri, 8 Aug 2025 11:27:22 +0200 Subject: [PATCH 04/16] Update docs/dqx/docs/demos.mdx --- docs/dqx/docs/demos.mdx | 1 - 1 file changed, 1 deletion(-) diff --git a/docs/dqx/docs/demos.mdx b/docs/dqx/docs/demos.mdx index 8da9bf74e..aebdc3c85 100644 --- a/docs/dqx/docs/demos.mdx +++ b/docs/dqx/docs/demos.mdx @@ -8,7 +8,6 @@ import Admonition from '@theme/Admonition'; Import the following notebooks in the Databricks workspace to try DQX out: ## Use as Library - * [DQX Quick Start Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_quick_start_demo_library.py) - quickstart on how to use DQX as a library. * [DQX Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_library.py) - demonstrates how to use DQX as a library. * [DQX Demo Notebook for Spark Structured Streaming](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo.py) - demonstrates how to use DQX as a library with Spark Structured Streaming. From 9cfcbb8b4484a8a90a383dc88ad9aebcbe3340f3 Mon Sep 17 00:00:00 2001 From: dinbab1984 Date: Sat, 9 Aug 2025 16:42:21 +0200 Subject: [PATCH 05/16] Issue#489 - Feature - DQX Spark Structured Streaming Demo --- ...ming_demo.py => dqx_streaming_demo_diy.py} | 75 +++-- demos/dqx_streaming_demo_native.py | 256 ++++++++++++++++++ docs/dqx/docs/demos.mdx | 3 +- tests/e2e/test_run_demos.py | 35 ++- 4 files changed, 346 insertions(+), 23 deletions(-) rename demos/{dqx_streaming_demo.py => dqx_streaming_demo_diy.py} (83%) create mode 100644 demos/dqx_streaming_demo_native.py diff --git a/demos/dqx_streaming_demo.py b/demos/dqx_streaming_demo_diy.py similarity index 83% rename from demos/dqx_streaming_demo.py rename to demos/dqx_streaming_demo_diy.py index 026bd2589..adc90383b 100644 --- a/demos/dqx_streaming_demo.py +++ b/demos/dqx_streaming_demo_diy.py @@ -52,21 +52,6 @@ # COMMAND ---------- -# MAGIC %md -# MAGIC ### Bronze Stream -# MAGIC -# MAGIC This cell reads the NYC Taxi 2019 dataset from a Delta table as a streaming DataFrame, which will be used as the input for downstream data quality checks and transformations. - -# COMMAND ---------- - -bronze_stream = ( - spark.readStream - .format("delta") - .load("/databricks-datasets/delta-sharing/samples/nyctaxi_2019") -) - -# COMMAND ---------- - # MAGIC %md # MAGIC ### Define Data Quality Checks # MAGIC @@ -148,6 +133,30 @@ # COMMAND ---------- +# MAGIC %md +# MAGIC +# MAGIC

Checkpoint and Table Location Options

+# MAGIC
    +# MAGIC
  • +# MAGIC Checkpoint Location: +# MAGIC
      +# MAGIC
    • Local path (e.g., /tmp/checkpoints/...)
      • +# MAGIC
    • Workspace path (e.g., /dbfs/mnt/...)
      • +# MAGIC
    • Unity Catalog (UC) volume path (e.g., /Volumes/catalog/schema/volume/...)
      • +# MAGIC
    +# MAGIC
    • +# MAGIC
  • +# MAGIC Silver and Quarantine Table: +# MAGIC
      +# MAGIC
    • Delta table path (e.g., /mnt/delta/silver_table)
      • +# MAGIC
    • Unity Catalog table (e.g., catalog.schema.table)
      • +# MAGIC
    +# MAGIC
    • +# MAGIC
+# MAGIC + +# COMMAND ---------- + dbutils.widgets.text("silver_checkpoint", "/tmp/dq/structured/bronze_to_silver_checkpoint") dbutils.widgets.text("quarantine_checkpoint", "/tmp/dq/structured/bronze_to_quarantine_checkpoint") dbutils.widgets.text("silver_table", "/tmp/dq/structured/silver_table") @@ -162,6 +171,21 @@ # COMMAND ---------- +# MAGIC %md +# MAGIC ### Bronze Stream +# MAGIC +# MAGIC This cell reads the NYC Taxi 2019 dataset from a Delta table as a streaming DataFrame, which will be used as the input for downstream data quality checks and transformations. + +# COMMAND ---------- + +bronze_stream = ( + spark.readStream + .format("delta") + .load("/databricks-datasets/delta-sharing/samples/nyctaxi_2019") +) + +# COMMAND ---------- + # MAGIC %md # MAGIC ### Apply data quality checks and split the bronze stream into silver and quarantine DataFrames @@ -173,10 +197,7 @@ dq_engine = DQEngine(WorkspaceClient()) -def apply_data_quality_and_split(df: DataFrame): - return dq_engine.apply_checks_by_metadata_and_split(df, checks) - -silver_df, quarantine_df = apply_data_quality_and_split(bronze_stream) +silver_df, quarantine_df = dq_engine.apply_checks_by_metadata_and_split(bronze_stream, checks) # COMMAND ---------- @@ -211,4 +232,18 @@ def write_stream(df, checkpoint_location, target): # COMMAND ---------- silver_query.awaitTermination() -quarantine_query.awaitTermination() \ No newline at end of file +quarantine_query.awaitTermination() + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ### Demo Clearup + +# COMMAND ---------- + +dbutils.fs.rm(silver_checkpoint, True) +dbutils.fs.rm(quarantine_checkpoint, True) +dbutils.fs.rm(silver_table, True) +dbutils.fs.rm(quarantine_table, True) +spark.sql("DROP TABLE IF EXISTS silver_table") +spark.sql("DROP TABLE IF EXISTS quarantine_table") \ No newline at end of file diff --git a/demos/dqx_streaming_demo_native.py b/demos/dqx_streaming_demo_native.py new file mode 100644 index 000000000..9629fa0bc --- /dev/null +++ b/demos/dqx_streaming_demo_native.py @@ -0,0 +1,256 @@ +# Databricks notebook source +# MAGIC +# MAGIC %md +# MAGIC ### Environment Setup for DQX with Structured Streaming +# MAGIC +# MAGIC #### 1\. Library Installation +# MAGIC +# MAGIC **Option A — Notebook-scoped installation** +# MAGIC +# MAGIC - Use notebook-scoped pip installs, which take effect only for the current interactive notebook session. +# MAGIC +# MAGIC ```py +# MAGIC # Install DQX library in the current notebook session +# MAGIC # Run this cell before any imports from databricks-labs-dqx +# MAGIC +# MAGIC %pip install databricks-labs-dqx +# MAGIC ``` +# MAGIC +# MAGIC - If you are using a cluster that was running prior to this install, you may need to restart the Python process for the changes to take full effect: +# MAGIC +# MAGIC ```py +# MAGIC # Optional: Force restart of the Python process if required by your environment +# MAGIC dbutils.library.restartPython() +# MAGIC ``` +# MAGIC +# MAGIC **Option B — Cluster-scoped installation (recommended for production streaming jobs)** +# MAGIC +# MAGIC - Install the DQX library to your cluster via the Databricks Libraries UI or by specifying it as a PyPI library dependency during cluster creation. +# MAGIC - This ensures the package is available for all jobs or notebooks attached to the cluster, including those running streaming workloads. +# MAGIC - The syntax for specifying the package is: +# MAGIC - Library Source: PyPI +# MAGIC - Package: `databricks-labs-dqx` +# MAGIC + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ### Install DQX as Library
+# MAGIC For this demo, we will install DQX as library +# MAGIC + +# COMMAND ---------- + +dbutils.widgets.text("test_library_ref", "", "Test Library Ref") + +if dbutils.widgets.get("test_library_ref") != "": + %pip install '{dbutils.widgets.get("test_library_ref")}' +else: + %pip install databricks-labs-dqx + +%restart_python + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ### Define Data Quality Checks +# MAGIC +# MAGIC This cell defines a set of data quality checks using YAML syntax. The checks are designed to validate key columns in the NYC Taxi dataset, such as `vendor_id`, `pickup_datetime`, `dropoff_datetime`, `passenger_count`, and `trip_distance`. Each check specifies a function, its arguments, a name, and a criticality level (error or warn). These checks will be used by the DQEngine to enforce data quality rules on the streaming data. + +# COMMAND ---------- + +# Define Data Quality checks +import yaml + + +checks = yaml.safe_load(""" +- check: + function: is_not_null + arguments: + column: vendor_id + name: vendor_id_is_null + criticality: error +- check: + function: is_not_null_and_not_empty + arguments: + column: vendor_id + trim_strings: true + name: vendor_id_is_null_or_empty + criticality: error + +- check: + function: is_not_null + arguments: + column: pickup_datetime + name: pickup_datetime_is_null + criticality: error +- check: + function: is_not_in_future + arguments: + column: pickup_datetime + name: pickup_datetime_isnt_in_range + criticality: warn + +- check: + function: is_not_in_future + arguments: + column: pickup_datetime + name: pickup_datetime_not_in_future + criticality: warn +- check: + function: is_not_in_future + arguments: + column: dropoff_datetime + name: dropoff_datetime_not_in_future + criticality: warn +- check: + function: is_not_null + arguments: + column: passenger_count + name: passenger_count_is_null + criticality: error +- check: + function: is_in_range + arguments: + column: passenger_count + min_limit: 0 + max_limit: 6 + name: passenger_incorrect_count + criticality: warn +- check: + function: is_not_null + arguments: + column: trip_distance + name: trip_distance_is_null + criticality: error +""") + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ### Inputs +# MAGIC - Checkpoint and table locations are set via widgets + +# COMMAND ---------- + +# MAGIC %md +# MAGIC +# MAGIC

Checkpoint and Table Location Options

+# MAGIC
    +# MAGIC
  • +# MAGIC Checkpoint Location: +# MAGIC
      +# MAGIC
    • Local path (e.g., /tmp/checkpoints/...)
      • +# MAGIC
    • Workspace path (e.g., /dbfs/mnt/...)
      • +# MAGIC
    • Unity Catalog (UC) volume path (e.g., /Volumes/catalog/schema/volume/...)
      • +# MAGIC
    +# MAGIC
    • +# MAGIC
  • +# MAGIC Silver and Quarantine Table: +# MAGIC
      +# MAGIC
    • Unity Catalog table (e.g., catalog.schema.table)
      • +# MAGIC
    +# MAGIC
    • +# MAGIC
+# MAGIC + +# COMMAND ---------- + +# DBTITLE 1,Set Catalog, Schema & checkpoint location for Demo Dataset +default_catalog_name = "main" +default_schema_name = "default" + +dbutils.widgets.text("demo_catalog_name", default_catalog_name, "Catalog Name") +dbutils.widgets.text("demo_schema_name", default_schema_name, "Schema Name") + +dbutils.widgets.text("silver_checkpoint", "/tmp/dq/structured/bronze_to_silver_checkpoint") +dbutils.widgets.text("quarantine_checkpoint", "/tmp/dq/structured/bronze_to_quarantine_checkpoint") + +# COMMAND ---------- + +from uuid import uuid4 + +catalog = dbutils.widgets.get("demo_catalog_name") +schema = dbutils.widgets.get("demo_schema_name") + +print(f"Selected Catalog for Demo Dataset: {catalog}") +print(f"Selected Schema for Demo Dataset: {schema}") + +silver_checkpoint = dbutils.widgets.get("silver_checkpoint") +quarantine_checkpoint = dbutils.widgets.get("quarantine_checkpoint") +uuid = uuid4() +silver_table = f"{catalog}.{schema}.`silver_{uuid}`" +quarantine_table = f"{catalog}.{schema}.`quarantine_{uuid}`" + +print(f"Demo Silver Table: {silver_table}") +print(f"Demo Quarantine Table: {quarantine_table}") + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ### Apply data quality checks to streaming daat using DQEngine Native method +# MAGIC This script performs data quality checks on a Parquet dataset using Databricks Labs DQX. +# MAGIC +# MAGIC 1. Reads a sample NYC Taxi dataset and writes it as Parquet files to a bronze location. This is needed for a sample parquet file(s) for streaming ingestion. +# MAGIC 2. Configures input for streaming ingestion using Auto Loader (cloudFiles) with schema tracking. +# MAGIC 3. Sets up output configurations for both the silver table and a quarantine table, specifying Delta format, checkpoint locations, and schema merging. +# MAGIC 4. Instantiates the DQEngine. +# MAGIC 5. Using DQEngine native method `apply_checks_by_metadata_and_save_in_table`, Applies data quality checks (defined in `checks`) to the input data stream, saving valid records to the silver table and invalid records to the quarantine table. + +# COMMAND ---------- + +from databricks.labs.dqx.engine import DQEngine +from databricks.labs.dqx.config import InputConfig, OutputConfig +from databricks.sdk import WorkspaceClient +from pyspark.sql import DataFrame + +bronze_loc = "/tmp/dq/bronze" +bronze_df = spark.read.load("/databricks-datasets/delta-sharing/samples/nyctaxi_2019") +bronze_df.write.mode("overwrite").format("parquet").save(f"{bronze_loc}/data") + +input_config=InputConfig( + location=f"{bronze_loc}/data", + format="cloudFiles", + options={"cloudFiles.format": "parquet", "cloudFiles.schemaLocation": f"{bronze_loc}/schema"}, + is_streaming=True +) + +output_config=OutputConfig( + location=silver_table, + format="delta", + mode="append", + trigger={"availableNow": True}, + options={"checkpointLocation": f"{silver_checkpoint}", "mergeSchema": "true"} +) + +quarantine_config=OutputConfig( + location=quarantine_table, + format="delta", + mode="append", + trigger={"availableNow": True}, + options={"checkpointLocation": f"{quarantine_checkpoint}", "mergeSchema": "true"} +) + + +dq_engine = DQEngine(WorkspaceClient()) + +dq_engine.apply_checks_by_metadata_and_save_in_table( + checks=checks, + input_config=input_config, + output_config=output_config, + quarantine_config=quarantine_config +) + + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ### Demo Cleanup + +# COMMAND ---------- + +dbutils.fs.rm(bronze_loc, True) +dbutils.fs.rm(silver_checkpoint, True) +dbutils.fs.rm(quarantine_checkpoint, True) +spark.sql(f"DROP TABLE IF EXISTS {silver_table}") +spark.sql(f"DROP TABLE IF EXISTS {quarantine_table}") \ No newline at end of file diff --git a/docs/dqx/docs/demos.mdx b/docs/dqx/docs/demos.mdx index aebdc3c85..71362a79b 100644 --- a/docs/dqx/docs/demos.mdx +++ b/docs/dqx/docs/demos.mdx @@ -10,7 +10,8 @@ Import the following notebooks in the Databricks workspace to try DQX out: ## Use as Library * [DQX Quick Start Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_quick_start_demo_library.py) - quickstart on how to use DQX as a library. * [DQX Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_library.py) - demonstrates how to use DQX as a library. -* [DQX Demo Notebook for Spark Structured Streaming](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo.py) - demonstrates how to use DQX as a library with Spark Structured Streaming. +* [DQX Demo (Native End-to-End Approach) Notebook for Spark Structured Streaming](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_native.py) - demonstrates how to use DQX as a library with Spark Structured Streaming (using All-in-one native method `apply_checks_by_metadata_and_save_in_table`; handles read-process-write internally). +* [DQX Demo (DIY Approach) Notebook for Spark Structured Streaming](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_diy.py) - demonstrates how to use DQX as a library with Spark Structured Streaming (Writing your own Streaming Pipeline: read → `apply_checks_by_metadata_and_split` → write). * [DQX Demo Notebook for Lakeflow Pipelines (DLT)](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_dlt_demo.py) - demonstrates how to use DQX as a library with Lakeflow Pipelines (formerly Delta Live Tables (DLT)). * [DQX Asset Bundles Demo](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_asset_bundle/README.md) - demonstrates how to use DQX as a library with Databricks Asset Bundles. * [DQX Demo for dbt](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_dbt/README.md) - demonstrates how to use DQX as a library with dbt projects. diff --git a/tests/e2e/test_run_demos.py b/tests/e2e/test_run_demos.py index 1ded8fae1..e6a86539a 100644 --- a/tests/e2e/test_run_demos.py +++ b/tests/e2e/test_run_demos.py @@ -194,10 +194,41 @@ def validate_demo_run_status(run: Run, client: WorkspaceClient) -> None: ), f"Run of '{task.task_key}' failed with message: {client.jobs.get_run_output(task.run_id).error}" -def test_run_dqx_streaming_demo(make_notebook, make_job, tmp_path): +def test_run_dqx_streaming_demo_native(make_notebook, make_schema, make_job, tmp_path): ws = WorkspaceClient() - path = Path(__file__).parent.parent.parent / "demos" / "dqx_streaming_demo.py" + path = Path(__file__).parent.parent.parent / "demos" / "dqx_streaming_demo_native.py" + with open(path, "rb") as f: + notebook = make_notebook(content=f, format=ImportFormat.SOURCE) + catalog = "main" + schema = make_schema(catalog_name=catalog).name + notebook_path = notebook.as_fuse().as_posix() + + # Use the temporary directory for outputs + run_id = str(uuid4()) + base_output_path = tmp_path / run_id + base_parameters = { + "demo_catalog_name": catalog, + "demo_schema_name": schema, + "silver_checkpoint": f"{base_output_path}/silver_checkpoint", + "quarantine_checkpoint": f"{base_output_path}/quarantine_checkpoint", + "test_library_ref": TEST_LIBRARY_REF, + } + notebook_task = NotebookTask(notebook_path=notebook_path, base_parameters=base_parameters) + job = make_job(tasks=[Task(task_key="dqx_streaming_demo", notebook_task=notebook_task)]) + waiter = ws.jobs.run_now_and_wait(job.job_id) + run = ws.jobs.wait_get_run_job_terminated_or_skipped( + run_id=waiter.run_id, + timeout=timedelta(minutes=30), + callback=lambda r: validate_demo_run_status(r, client=ws), + ) + logging.info(f"Job run {run.run_id} completed successfully for dqx_streaming_demo") + + +def test_run_dqx_streaming_demo_diy(make_notebook, make_job, tmp_path): + + ws = WorkspaceClient() + path = Path(__file__).parent.parent.parent / "demos" / "dqx_streaming_demo_diy.py" with open(path, "rb") as f: notebook = make_notebook(content=f, format=ImportFormat.SOURCE) notebook_path = notebook.as_fuse().as_posix() From 5439a2a2df44402a697e9821fcdce2720da004f7 Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Sun, 10 Aug 2025 10:46:53 +0200 Subject: [PATCH 06/16] Update demos/dqx_streaming_demo_native.py --- demos/dqx_streaming_demo_native.py | 1 - 1 file changed, 1 deletion(-) diff --git a/demos/dqx_streaming_demo_native.py b/demos/dqx_streaming_demo_native.py index 9629fa0bc..9dc433db5 100644 --- a/demos/dqx_streaming_demo_native.py +++ b/demos/dqx_streaming_demo_native.py @@ -241,7 +241,6 @@ quarantine_config=quarantine_config ) - # COMMAND ---------- # MAGIC %md From 284f8ab0c907e039ceca3294207f76238dcc8ccd Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Sun, 10 Aug 2025 10:46:59 +0200 Subject: [PATCH 07/16] Update demos/dqx_streaming_demo_native.py --- demos/dqx_streaming_demo_native.py | 1 - 1 file changed, 1 deletion(-) diff --git a/demos/dqx_streaming_demo_native.py b/demos/dqx_streaming_demo_native.py index 9dc433db5..5459a7aa2 100644 --- a/demos/dqx_streaming_demo_native.py +++ b/demos/dqx_streaming_demo_native.py @@ -231,7 +231,6 @@ options={"checkpointLocation": f"{quarantine_checkpoint}", "mergeSchema": "true"} ) - dq_engine = DQEngine(WorkspaceClient()) dq_engine.apply_checks_by_metadata_and_save_in_table( From 5a4096e4c1386ab430cb0492d3278d2933189e33 Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Sun, 10 Aug 2025 10:47:05 +0200 Subject: [PATCH 08/16] Update demos/dqx_streaming_demo_native.py --- demos/dqx_streaming_demo_native.py | 1 + 1 file changed, 1 insertion(+) diff --git a/demos/dqx_streaming_demo_native.py b/demos/dqx_streaming_demo_native.py index 5459a7aa2..55f0c1324 100644 --- a/demos/dqx_streaming_demo_native.py +++ b/demos/dqx_streaming_demo_native.py @@ -204,6 +204,7 @@ from databricks.sdk import WorkspaceClient from pyspark.sql import DataFrame +# prepare sample test data bronze_loc = "/tmp/dq/bronze" bronze_df = spark.read.load("/databricks-datasets/delta-sharing/samples/nyctaxi_2019") bronze_df.write.mode("overwrite").format("parquet").save(f"{bronze_loc}/data") From b69dc60f3db958189f1289b54f5a3ab7a783d6a0 Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Sun, 10 Aug 2025 10:47:12 +0200 Subject: [PATCH 09/16] Update demos/dqx_streaming_demo_native.py --- demos/dqx_streaming_demo_native.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/demos/dqx_streaming_demo_native.py b/demos/dqx_streaming_demo_native.py index 55f0c1324..637db2a64 100644 --- a/demos/dqx_streaming_demo_native.py +++ b/demos/dqx_streaming_demo_native.py @@ -195,7 +195,7 @@ # MAGIC 2. Configures input for streaming ingestion using Auto Loader (cloudFiles) with schema tracking. # MAGIC 3. Sets up output configurations for both the silver table and a quarantine table, specifying Delta format, checkpoint locations, and schema merging. # MAGIC 4. Instantiates the DQEngine. -# MAGIC 5. Using DQEngine native method `apply_checks_by_metadata_and_save_in_table`, Applies data quality checks (defined in `checks`) to the input data stream, saving valid records to the silver table and invalid records to the quarantine table. +# MAGIC 5. Using `DQEngine` native method `apply_checks_by_metadata_and_save_in_table`, applies data quality checks (defined in `checks`) to the input data stream, saving valid records to the silver table and invalid records to the quarantine table. # COMMAND ---------- From 528924418439d94586b619919e2edc5a9f4c159e Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Sun, 10 Aug 2025 10:47:18 +0200 Subject: [PATCH 10/16] Update demos/dqx_streaming_demo_native.py --- demos/dqx_streaming_demo_native.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/demos/dqx_streaming_demo_native.py b/demos/dqx_streaming_demo_native.py index 637db2a64..2087e1faa 100644 --- a/demos/dqx_streaming_demo_native.py +++ b/demos/dqx_streaming_demo_native.py @@ -188,7 +188,7 @@ # COMMAND ---------- # MAGIC %md -# MAGIC ### Apply data quality checks to streaming daat using DQEngine Native method +# MAGIC ### Apply data quality checks to streaming data using `DQEngine` Native method # MAGIC This script performs data quality checks on a Parquet dataset using Databricks Labs DQX. # MAGIC # MAGIC 1. Reads a sample NYC Taxi dataset and writes it as Parquet files to a bronze location. This is needed for a sample parquet file(s) for streaming ingestion. From 20dc407d42508848a2f7dacc1fa12c9a94c94413 Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Sun, 10 Aug 2025 10:49:11 +0200 Subject: [PATCH 11/16] Update demos/dqx_streaming_demo_diy.py --- demos/dqx_streaming_demo_diy.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/demos/dqx_streaming_demo_diy.py b/demos/dqx_streaming_demo_diy.py index adc90383b..688f88b50 100644 --- a/demos/dqx_streaming_demo_diy.py +++ b/demos/dqx_streaming_demo_diy.py @@ -214,7 +214,7 @@ def write_stream(df, checkpoint_location, target): .format("delta") .outputMode("append") .option("checkpointLocation", checkpoint_location) - .trigger(availableNow=True) + .trigger(availableNow=True) # stop the stream as soon as bronze data is processed ) if target.startswith(PATH_PREFIXES): return writer.start(target) From 4235d25806fe2955e6f30333c48a75159b8541ba Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Sun, 10 Aug 2025 10:50:53 +0200 Subject: [PATCH 12/16] Update docs/dqx/docs/demos.mdx --- docs/dqx/docs/demos.mdx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/dqx/docs/demos.mdx b/docs/dqx/docs/demos.mdx index 71362a79b..170669cbd 100644 --- a/docs/dqx/docs/demos.mdx +++ b/docs/dqx/docs/demos.mdx @@ -10,7 +10,7 @@ Import the following notebooks in the Databricks workspace to try DQX out: ## Use as Library * [DQX Quick Start Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_quick_start_demo_library.py) - quickstart on how to use DQX as a library. * [DQX Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_library.py) - demonstrates how to use DQX as a library. -* [DQX Demo (Native End-to-End Approach) Notebook for Spark Structured Streaming](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_native.py) - demonstrates how to use DQX as a library with Spark Structured Streaming (using All-in-one native method `apply_checks_by_metadata_and_save_in_table`; handles read-process-write internally). +* [DQX Demo Notebook for Spark Structured Streaming (Native End-to-End Approach)](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_native.py) - demonstrates how to use DQX as a library with Spark Structured Streaming (using all-in-one native method `apply_checks_by_metadata_and_save_in_table`). * [DQX Demo (DIY Approach) Notebook for Spark Structured Streaming](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_diy.py) - demonstrates how to use DQX as a library with Spark Structured Streaming (Writing your own Streaming Pipeline: read → `apply_checks_by_metadata_and_split` → write). * [DQX Demo Notebook for Lakeflow Pipelines (DLT)](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_dlt_demo.py) - demonstrates how to use DQX as a library with Lakeflow Pipelines (formerly Delta Live Tables (DLT)). * [DQX Asset Bundles Demo](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_asset_bundle/README.md) - demonstrates how to use DQX as a library with Databricks Asset Bundles. From 42832820a9bb90d6dc4bf61a0f4af280c9809d26 Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Sun, 10 Aug 2025 10:51:08 +0200 Subject: [PATCH 13/16] Update docs/dqx/docs/demos.mdx --- docs/dqx/docs/demos.mdx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/dqx/docs/demos.mdx b/docs/dqx/docs/demos.mdx index 170669cbd..f6c14a9f9 100644 --- a/docs/dqx/docs/demos.mdx +++ b/docs/dqx/docs/demos.mdx @@ -11,7 +11,7 @@ Import the following notebooks in the Databricks workspace to try DQX out: * [DQX Quick Start Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_quick_start_demo_library.py) - quickstart on how to use DQX as a library. * [DQX Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_library.py) - demonstrates how to use DQX as a library. * [DQX Demo Notebook for Spark Structured Streaming (Native End-to-End Approach)](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_native.py) - demonstrates how to use DQX as a library with Spark Structured Streaming (using all-in-one native method `apply_checks_by_metadata_and_save_in_table`). -* [DQX Demo (DIY Approach) Notebook for Spark Structured Streaming](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_diy.py) - demonstrates how to use DQX as a library with Spark Structured Streaming (Writing your own Streaming Pipeline: read → `apply_checks_by_metadata_and_split` → write). +* [DQX Demo Notebook for Spark Structured Streaming (DIY Approach)](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_diy.py) - demonstrates how to use DQX as a library with Spark Structured Streaming (writing your own Streaming Pipeline: read → `apply_checks_by_metadata_and_split` → write). * [DQX Demo Notebook for Lakeflow Pipelines (DLT)](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_dlt_demo.py) - demonstrates how to use DQX as a library with Lakeflow Pipelines (formerly Delta Live Tables (DLT)). * [DQX Asset Bundles Demo](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_asset_bundle/README.md) - demonstrates how to use DQX as a library with Databricks Asset Bundles. * [DQX Demo for dbt](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_dbt/README.md) - demonstrates how to use DQX as a library with dbt projects. From c8c073ed5e7a9606128ab752adb2bd34de8d4b2d Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Tue, 12 Aug 2025 23:18:00 +0200 Subject: [PATCH 14/16] updated docs --- demos/dqx_streaming_demo_diy.py | 49 +++++++++++++------------- demos/dqx_streaming_demo_native.py | 55 ++++++++++++++---------------- docs/dqx/docs/demos.mdx | 4 +-- 3 files changed, 51 insertions(+), 57 deletions(-) diff --git a/demos/dqx_streaming_demo_diy.py b/demos/dqx_streaming_demo_diy.py index 688f88b50..862123bee 100644 --- a/demos/dqx_streaming_demo_diy.py +++ b/demos/dqx_streaming_demo_diy.py @@ -1,5 +1,4 @@ # Databricks notebook source -# MAGIC # MAGIC %md # MAGIC ### Environment Setup for DQX with Structured Streaming # MAGIC @@ -7,30 +6,20 @@ # MAGIC # MAGIC **Option A — Notebook-scoped installation** # MAGIC -# MAGIC - Use notebook-scoped pip installs, which take effect only for the current interactive notebook session. -# MAGIC -# MAGIC ```py -# MAGIC # Install DQX library in the current notebook session -# MAGIC # Run this cell before any imports from databricks-labs-dqx +# MAGIC - Use notebook-scoped pip installs, which apply only to the current interactive notebook session: # MAGIC -# MAGIC %pip install databricks-labs-dqx -# MAGIC ``` +# MAGIC `%pip install databricks-labs-dqx` # MAGIC -# MAGIC - If you are using a cluster that was running prior to this install, you may need to restart the Python process for the changes to take full effect: +# MAGIC - If your cluster was running before this installation, you may need to restart the Python process for the changes to take effect: # MAGIC -# MAGIC ```py -# MAGIC # Optional: Force restart of the Python process if required by your environment -# MAGIC dbutils.library.restartPython() -# MAGIC ``` +# MAGIC `dbutils.library.restartPython()` # MAGIC # MAGIC **Option B — Cluster-scoped installation (recommended for production streaming jobs)** # MAGIC -# MAGIC - Install the DQX library to your cluster via the Databricks Libraries UI or by specifying it as a PyPI library dependency during cluster creation. -# MAGIC - This ensures the package is available for all jobs or notebooks attached to the cluster, including those running streaming workloads. -# MAGIC - The syntax for specifying the package is: -# MAGIC - Library Source: PyPI +# MAGIC - Install the DQX library on your cluster via the Databricks Libraries UI or by adding it as a PyPI dependency when creating the cluster. This makes the package available to all jobs and notebooks attached to the cluster, including streaming workloads. +# MAGIC - To specify the package use the following: +# MAGIC - Library Source: PyPI # MAGIC - Package: `databricks-labs-dqx` -# MAGIC # COMMAND ---------- @@ -55,7 +44,7 @@ # MAGIC %md # MAGIC ### Define Data Quality Checks # MAGIC -# MAGIC This cell defines a set of data quality checks using YAML syntax. The checks are designed to validate key columns in the NYC Taxi dataset, such as `vendor_id`, `pickup_datetime`, `dropoff_datetime`, `passenger_count`, and `trip_distance`. Each check specifies a function, its arguments, a name, and a criticality level (error or warn). These checks will be used by the DQEngine to enforce data quality rules on the streaming data. +# MAGIC This cell defines a set of data quality checks using YAML syntax. The checks are designed to validate key columns in the NYC Taxi dataset, such as `vendor_id`, `pickup_datetime`, `dropoff_datetime`, `passenger_count`, and `trip_distance`. Each check specifies a function, its arguments, a name, and a criticality level (`error` or `warn`). These checks will be used by the `DQEngine` to enforce data quality rules on the streaming data. # COMMAND ---------- @@ -129,7 +118,7 @@ # MAGIC %md # MAGIC ### Inputs -# MAGIC - Checkpoint and table locations are set via widgets +# MAGIC Checkpoint and table locations are set via widgets. # COMMAND ---------- @@ -138,7 +127,7 @@ # MAGIC

Checkpoint and Table Location Options

# MAGIC
    # MAGIC
  • -# MAGIC Checkpoint Location: +# MAGIC Checkpoint Location can be specified as: # MAGIC
      # MAGIC
    • Local path (e.g., /tmp/checkpoints/...)
      • # MAGIC
    • Workspace path (e.g., /dbfs/mnt/...)
      • @@ -148,12 +137,10 @@ # MAGIC
    • # MAGIC Silver and Quarantine Table: # MAGIC
        -# MAGIC
      • Delta table path (e.g., /mnt/delta/silver_table)
        • # MAGIC
      • Unity Catalog table (e.g., catalog.schema.table)
        • # MAGIC
      # MAGIC
      • # MAGIC
    -# MAGIC # COMMAND ---------- @@ -187,7 +174,7 @@ # COMMAND ---------- # MAGIC %md -# MAGIC ### Apply data quality checks and split the bronze stream into silver and quarantine DataFrames +# MAGIC ### Apply data quality checks and split the bronze data into silver and quarantine DataFrames # COMMAND ---------- @@ -214,7 +201,7 @@ def write_stream(df, checkpoint_location, target): .format("delta") .outputMode("append") .option("checkpointLocation", checkpoint_location) - .trigger(availableNow=True) # stop the stream as soon as bronze data is processed + .trigger(availableNow=True) # stop the stream once all data is processed ) if target.startswith(PATH_PREFIXES): return writer.start(target) @@ -227,7 +214,7 @@ def write_stream(df, checkpoint_location, target): # COMMAND ---------- # MAGIC %md -# MAGIC ### Wait for the streams to finish or keep them running for interactive sessions if required +# MAGIC ### Wait for the streams to finish or keep them running for interactive sessions # COMMAND ---------- @@ -236,6 +223,16 @@ def write_stream(df, checkpoint_location, target): # COMMAND ---------- +# MAGIC %md +# MAGIC ### Display Results + +# COMMAND ---------- + +display(spark.sql(f"SELECT * FROM delta.`{silver_table}`")) +display(spark.sql(f"SELECT * FROM delta.`{quarantine_table}`")) + +# COMMAND ---------- + # MAGIC %md # MAGIC ### Demo Clearup diff --git a/demos/dqx_streaming_demo_native.py b/demos/dqx_streaming_demo_native.py index 2087e1faa..d58fda83f 100644 --- a/demos/dqx_streaming_demo_native.py +++ b/demos/dqx_streaming_demo_native.py @@ -1,5 +1,4 @@ # Databricks notebook source -# MAGIC # MAGIC %md # MAGIC ### Environment Setup for DQX with Structured Streaming # MAGIC @@ -7,36 +6,26 @@ # MAGIC # MAGIC **Option A — Notebook-scoped installation** # MAGIC -# MAGIC - Use notebook-scoped pip installs, which take effect only for the current interactive notebook session. -# MAGIC -# MAGIC ```py -# MAGIC # Install DQX library in the current notebook session -# MAGIC # Run this cell before any imports from databricks-labs-dqx +# MAGIC - Use notebook-scoped pip installs, which apply only to the current interactive notebook session: # MAGIC -# MAGIC %pip install databricks-labs-dqx -# MAGIC ``` +# MAGIC `%pip install databricks-labs-dqx` # MAGIC -# MAGIC - If you are using a cluster that was running prior to this install, you may need to restart the Python process for the changes to take full effect: +# MAGIC - If your cluster was running before this installation, you may need to restart the Python process for the changes to take effect: # MAGIC -# MAGIC ```py -# MAGIC # Optional: Force restart of the Python process if required by your environment -# MAGIC dbutils.library.restartPython() -# MAGIC ``` +# MAGIC `dbutils.library.restartPython()` # MAGIC # MAGIC **Option B — Cluster-scoped installation (recommended for production streaming jobs)** # MAGIC -# MAGIC - Install the DQX library to your cluster via the Databricks Libraries UI or by specifying it as a PyPI library dependency during cluster creation. -# MAGIC - This ensures the package is available for all jobs or notebooks attached to the cluster, including those running streaming workloads. -# MAGIC - The syntax for specifying the package is: -# MAGIC - Library Source: PyPI +# MAGIC - Install the DQX library on your cluster via the Databricks Libraries UI or by adding it as a PyPI dependency when creating the cluster. This makes the package available to all jobs and notebooks attached to the cluster, including streaming workloads. +# MAGIC - To specify the package use the following: +# MAGIC - Library Source: PyPI # MAGIC - Package: `databricks-labs-dqx` -# MAGIC # COMMAND ---------- # MAGIC %md # MAGIC ### Install DQX as Library
    -# MAGIC For this demo, we will install DQX as library +# MAGIC For this demo, we will install DQX as library. # MAGIC # COMMAND ---------- @@ -55,7 +44,7 @@ # MAGIC %md # MAGIC ### Define Data Quality Checks # MAGIC -# MAGIC This cell defines a set of data quality checks using YAML syntax. The checks are designed to validate key columns in the NYC Taxi dataset, such as `vendor_id`, `pickup_datetime`, `dropoff_datetime`, `passenger_count`, and `trip_distance`. Each check specifies a function, its arguments, a name, and a criticality level (error or warn). These checks will be used by the DQEngine to enforce data quality rules on the streaming data. +# MAGIC This cell defines a set of data quality checks using YAML syntax. The checks are designed to validate key columns in the NYC Taxi dataset, such as `vendor_id`, `pickup_datetime`, `dropoff_datetime`, `passenger_count`, and `trip_distance`. Each check specifies a function, its arguments, a name, and a criticality level (`error` or `warn`). These checks will be used by the `DQEngine` to enforce data quality rules on the streaming data. # COMMAND ---------- @@ -129,7 +118,7 @@ # MAGIC %md # MAGIC ### Inputs -# MAGIC - Checkpoint and table locations are set via widgets +# MAGIC Checkpoint and table locations are set via widgets. # COMMAND ---------- @@ -138,7 +127,7 @@ # MAGIC

    Checkpoint and Table Location Options

    # MAGIC
      # MAGIC
    • -# MAGIC Checkpoint Location: +# MAGIC Checkpoint Location can be specified as: # MAGIC
        # MAGIC
      • Local path (e.g., /tmp/checkpoints/...)
        • # MAGIC
      • Workspace path (e.g., /dbfs/mnt/...)
        • @@ -152,7 +141,6 @@ # MAGIC
      # MAGIC
      • # MAGIC
    -# MAGIC # COMMAND ---------- @@ -189,13 +177,12 @@ # MAGIC %md # MAGIC ### Apply data quality checks to streaming data using `DQEngine` Native method -# MAGIC This script performs data quality checks on a Parquet dataset using Databricks Labs DQX. -# MAGIC +# MAGIC This script performs data quality checks on a Parquet dataset, as follows: # MAGIC 1. Reads a sample NYC Taxi dataset and writes it as Parquet files to a bronze location. This is needed for a sample parquet file(s) for streaming ingestion. # MAGIC 2. Configures input for streaming ingestion using Auto Loader (cloudFiles) with schema tracking. # MAGIC 3. Sets up output configurations for both the silver table and a quarantine table, specifying Delta format, checkpoint locations, and schema merging. -# MAGIC 4. Instantiates the DQEngine. -# MAGIC 5. Using `DQEngine` native method `apply_checks_by_metadata_and_save_in_table`, applies data quality checks (defined in `checks`) to the input data stream, saving valid records to the silver table and invalid records to the quarantine table. +# MAGIC 4. Instantiates the `DQEngine`. +# MAGIC 5. Using `DQEngine` native method `apply_checks_by_metadata_and_save_in_table`, applies data quality checks to the input data stream, saving valid records to the silver table and invalid records to the quarantine table. # COMMAND ---------- @@ -220,7 +207,7 @@ location=silver_table, format="delta", mode="append", - trigger={"availableNow": True}, + trigger={"availableNow": True}, # stop the stream once all data is processed options={"checkpointLocation": f"{silver_checkpoint}", "mergeSchema": "true"} ) @@ -228,7 +215,7 @@ location=quarantine_table, format="delta", mode="append", - trigger={"availableNow": True}, + trigger={"availableNow": True}, # stop the stream once all data is processed options={"checkpointLocation": f"{quarantine_checkpoint}", "mergeSchema": "true"} ) @@ -243,6 +230,16 @@ # COMMAND ---------- +# MAGIC %md +# MAGIC ### Display Results + +# COMMAND ---------- + +display(spark.sql(f"SELECT * FROM {silver_table}")) +display(spark.sql(f"SELECT * FROM {quarantine_table}")) + +# COMMAND ---------- + # MAGIC %md # MAGIC ### Demo Cleanup diff --git a/docs/dqx/docs/demos.mdx b/docs/dqx/docs/demos.mdx index f6c14a9f9..2ec8b2a31 100644 --- a/docs/dqx/docs/demos.mdx +++ b/docs/dqx/docs/demos.mdx @@ -10,8 +10,8 @@ Import the following notebooks in the Databricks workspace to try DQX out: ## Use as Library * [DQX Quick Start Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_quick_start_demo_library.py) - quickstart on how to use DQX as a library. * [DQX Demo Notebook](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_library.py) - demonstrates how to use DQX as a library. -* [DQX Demo Notebook for Spark Structured Streaming (Native End-to-End Approach)](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_native.py) - demonstrates how to use DQX as a library with Spark Structured Streaming (using all-in-one native method `apply_checks_by_metadata_and_save_in_table`). -* [DQX Demo Notebook for Spark Structured Streaming (DIY Approach)](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_diy.py) - demonstrates how to use DQX as a library with Spark Structured Streaming (writing your own Streaming Pipeline: read → `apply_checks_by_metadata_and_split` → write). +* [DQX Demo Notebook for Spark Structured Streaming (Native End-to-End Approach)](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_native.py) - demonstrates how to use DQX as a library with Spark Structured Streaming, using the built-in end-to-end method to handle both reading and writing. +* [DQX Demo Notebook for Spark Structured Streaming (DIY Approach)](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_streaming_demo_diy.py) - demonstrates how to use DQX as a library with Spark Structured Streaming, while handling reading and writing on your own outside DQX. * [DQX Demo Notebook for Lakeflow Pipelines (DLT)](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_dlt_demo.py) - demonstrates how to use DQX as a library with Lakeflow Pipelines (formerly Delta Live Tables (DLT)). * [DQX Asset Bundles Demo](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_asset_bundle/README.md) - demonstrates how to use DQX as a library with Databricks Asset Bundles. * [DQX Demo for dbt](https://github.com/databrickslabs/dqx/blob/v0.8.0/demos/dqx_demo_dbt/README.md) - demonstrates how to use DQX as a library with dbt projects. From e6bad71826cd9a70b2002848835b8c534b5078df Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Wed, 13 Aug 2025 00:07:21 +0200 Subject: [PATCH 15/16] test --- demos/dqx_streaming_demo_native.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/demos/dqx_streaming_demo_native.py b/demos/dqx_streaming_demo_native.py index d58fda83f..653b3408d 100644 --- a/demos/dqx_streaming_demo_native.py +++ b/demos/dqx_streaming_demo_native.py @@ -235,8 +235,8 @@ # COMMAND ---------- -display(spark.sql(f"SELECT * FROM {silver_table}")) -display(spark.sql(f"SELECT * FROM {quarantine_table}")) +#display(spark.sql(f"SELECT * FROM {silver_table}")) +#display(spark.sql(f"SELECT * FROM {quarantine_table}")) # COMMAND ---------- From 1551e34a51dc676c2bf99a941d0047424ba90e96 Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Wed, 13 Aug 2025 08:12:46 +0200 Subject: [PATCH 16/16] test --- demos/dqx_streaming_demo_native.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/demos/dqx_streaming_demo_native.py b/demos/dqx_streaming_demo_native.py index 653b3408d..5c7771323 100644 --- a/demos/dqx_streaming_demo_native.py +++ b/demos/dqx_streaming_demo_native.py @@ -1,4 +1,5 @@ # Databricks notebook source +# MAGIC # MAGIC %md # MAGIC ### Environment Setup for DQX with Structured Streaming # MAGIC @@ -235,8 +236,8 @@ # COMMAND ---------- -#display(spark.sql(f"SELECT * FROM {silver_table}")) -#display(spark.sql(f"SELECT * FROM {quarantine_table}")) +display(spark.sql(f"SELECT * FROM {silver_table}")) +display(spark.sql(f"SELECT * FROM {quarantine_table}")) # COMMAND ----------