diff --git a/AGENTS.md b/AGENTS.md index 441d87e74..da6701d5e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -113,7 +113,7 @@ tests/ - **Cover all changes with tests.** New check functions and rule logic → unit tests. Workspace interactions → integration tests. Bug fixes → regression tests. - **Unit tests** (`tests/unit/`) run without Spark or a live workspace and must stay fast. - **Integration tests** (`tests/integration/`) require a real workspace and spark session; do not add workspace API calls to unit tests. -- Test **behaviour, not implementation details**: assert on outputs and observable state, not on private methods or internal data structures. +- Test **behaviour, not implementation details**: assert on outputs and observable state, not on private methods or internal data structures. Do not call or access private/protected members (e.g. `obj._helper`) from tests — exercise the public API instead. If something is only reachable via a private member, that is a design smell: make it public or extract a shared helper, do not reach past the boundary (and never silence the resulting `protected-access` lint — see Critical Rule 6). - Use **dependency injection to enable testing**: construct dependencies with `create_autospec` rather than patching internal module state. - Use **pytest fixtures** (`conftest.py`) to share setup and teardown logic across tests. Unit-level fixtures live in `tests/unit/conftest.py`; integration-level fixtures in `tests/integration/conftest.py`. Do not duplicate fixture logic inline in individual tests. - For workspace resource creation and cleanup in integration tests, use the pytester `factory` helper — see [## Testing](#testing) for the established patterns. @@ -160,6 +160,8 @@ Use `ConfigSerializer` — it preserves nested types. `dataclasses.asdict()` los Fix the code instead of adding `# pylint: disable`, `# type: ignore`, `# noqa`, or per-file ignores. Use project-wide exceptions in `pyproject.toml` only when there is no viable fix (e.g., third-party API compatibility). +In particular, never disable `protected-access` (pylint `W0212`) — inline, per-file, or globally in `pyproject.toml` — to let tests or callers reach private/protected members (`_name`). That is a hack that masks a design smell and is technical debt. Instead, exercise the public API, or if a member genuinely needs outside access, make it public or extract a shared helper. + --- ## Security Requirements diff --git a/demos/dqx_dlt_demo.py b/demos/dqx_dlt_demo.py index a8464e9ed..25419646b 100644 --- a/demos/dqx_dlt_demo.py +++ b/demos/dqx_dlt_demo.py @@ -5,7 +5,12 @@ # MAGIC # MAGIC %md -# MAGIC ## Create Lakeflow Pipeline (formerly Delta Live Tables - DLT) +# MAGIC ## DQX in a Lakeflow Pipeline (formerly Delta Live Tables - DLT) +# MAGIC +# MAGIC This demo applies DQX checks and reports issues as additional columns (`_errors` / `_warnings`), +# MAGIC persisting the checked data as a `silver` table. Summary metrics are then computed as a materialized +# MAGIC view over that table. For a pipeline that quarantines invalid records into a separate table, see +# MAGIC `dqx_dlt_demo_quarantine.py`. # MAGIC # MAGIC Create new ETL Pipeline to execute this notebook (see [here](https://docs.databricks.com/aws/en/getting-started/data-pipeline-get-started)): # MAGIC 1. Upload the notebook to a Databricks Workspace @@ -23,27 +28,21 @@ # COMMAND ---------- -# MAGIC %md -# MAGIC ## Define Lakeflow Pipeline - -# COMMAND ---------- - +import yaml from databricks.labs.dqx.engine import DQEngine +from databricks.labs.dqx.metrics_observer import DQMetricsObserver from databricks.sdk import WorkspaceClient +# compute_summary_metrics requires an observer on the engine; it reads any custom_metrics from it. +dq_engine = DQEngine(WorkspaceClient(), observer=DQMetricsObserver()) + # COMMAND ---------- -@dlt.view -def bronze(): - df = spark.readStream.format("delta") \ - .load("/databricks-datasets/delta-sharing/samples/nyctaxi_2019") - return df +# MAGIC %md +# MAGIC ## Define Data Quality checks # COMMAND ---------- -# Define Data Quality checks -import yaml - # Define checks in YAML format. They can also be defined using classes or loaded from a file or table. checks = yaml.safe_load(""" - check: @@ -109,26 +108,36 @@ def bronze(): # COMMAND ---------- -dq_engine = DQEngine(WorkspaceClient()) +# MAGIC %md +# MAGIC ## Define Lakeflow Pipeline (bronze -> silver -> metrics) + +# COMMAND ---------- -# Read data from Bronze and apply checks +# Bronze: raw input as a streaming view. @dlt.view -def bronze_dq_check(): - df = dlt.read_stream("bronze") - return dq_engine.apply_checks_by_metadata(df, checks) +def bronze(): + return spark.readStream.format("delta") \ + .load("/databricks-datasets/delta-sharing/samples/nyctaxi_2019") # COMMAND ---------- -# # get rows without errors or warnings, and drop auxiliary columns +# Silver: apply checks and report issues as additional columns (_errors / _warnings). +# Persist as a table so it can be used downstream (including by the metrics view below). @dlt.table def silver(): - df = dlt.read_stream("bronze_dq_check") - return dq_engine.get_valid(df) + df = dlt.read_stream("bronze") + return dq_engine.apply_checks_by_metadata(df, checks) # COMMAND ---------- -# get only rows with errors or warnings +# Summary Metrics: materialized view computed by aggregation over the silver table. +# One row per metric (input / error / warning / valid row counts and per-check breakdown). +# Note: this MV is a cumulative snapshot over the whole table (input_row_count is the running +# total, not a per-run count). It refreshes incrementally only when the query is deterministic — +# set static run_time_overwrite / run_id_overwrite in ExtraParams for that. For per-run / per-window +# metrics on large or incrementally-checked tables, append from a windowed streaming table instead +# (see the "Snapshot vs. history" section of the Summary Metrics guide). @dlt.table -def quarantine(): - df = dlt.read_stream("bronze_dq_check") - return dq_engine.get_invalid(df) \ No newline at end of file +def dq_summary_metrics(): + df = dlt.read("silver") + return dq_engine.compute_summary_metrics(df, checks=checks) diff --git a/demos/dqx_dlt_demo_quarantine.py b/demos/dqx_dlt_demo_quarantine.py new file mode 100644 index 000000000..a7ffc998c --- /dev/null +++ b/demos/dqx_dlt_demo_quarantine.py @@ -0,0 +1,159 @@ +# Databricks notebook source +import dlt + +# COMMAND ---------- + +# MAGIC +# MAGIC %md +# MAGIC ## DQX in a Lakeflow Pipeline (formerly Delta Live Tables - DLT) — data quality with quarantine pattern +# MAGIC +# MAGIC This demo applies DQX checks and **splits** the data into a valid `silver` table and a `quarantine` +# MAGIC table. It also persists the checked layer (`bronze_dq_check`) so summary metrics can be computed as +# MAGIC a materialized view over all rows. For a simpler pipeline that reports issues as columns without +# MAGIC quarantining, see `dqx_dlt_demo.py`. +# MAGIC +# MAGIC Create new ETL Pipeline to execute this notebook (see [here](https://docs.databricks.com/aws/en/getting-started/data-pipeline-get-started)): +# MAGIC 1. Upload the notebook to a Databricks Workspace +# MAGIC 2. Go to `Workflows` tab > `Create` > `ETL Pipeline` > `Add existing assets` > select the source code path and root directory +# MAGIC 3. Add DQX library as a [dependency](https://docs.databricks.com/aws/en/dlt/dlt-multi-file-editor#environment) to the pipeline: Go to `Settings` > `Edit environment` > Add `databricks‑labs‑dqx` as dependency +# MAGIC 4. Run the pipeline +# MAGIC +# MAGIC +# MAGIC As an alternative to setting the environment as described above, you can also [install](https://docs.databricks.com/aws/en/dlt/external-dependencies) DQX directly in the notebook. Put the below commands as first cells in the notebook: +# MAGIC +# MAGIC %`pip install databricks-labs-dqx` +# MAGIC +# MAGIC `dbutils.library.restartPython()` +# MAGIC + +# COMMAND ---------- + +import yaml +from databricks.labs.dqx.engine import DQEngine +from databricks.labs.dqx.metrics_observer import DQMetricsObserver +from databricks.sdk import WorkspaceClient + +# compute_summary_metrics requires an observer on the engine; it reads any custom_metrics from it. +dq_engine = DQEngine(WorkspaceClient(), observer=DQMetricsObserver()) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Define Data Quality checks + +# COMMAND ---------- + +# Define checks in YAML format. They can also be defined using classes or loaded from a file or a table. +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 ## Define Lakeflow Pipeline (bronze -> silver + quarantine -> metrics) + +# COMMAND ---------- + +# Bronze: raw input as a streaming view. +@dlt.view +def bronze(): + return spark.readStream.format("delta") \ + .load("/databricks-datasets/delta-sharing/samples/nyctaxi_2019") + +# COMMAND ---------- + +# Apply checks and persist the checked data as a table so the summary-metrics materialized view +# below can read the result columns (_errors/_warnings) after the pipeline writes them. +@dlt.table +def bronze_dq_check(): + df = dlt.read_stream("bronze") + return dq_engine.apply_checks_by_metadata(df, checks) + +# COMMAND ---------- + +# Silver: rows without errors or warnings, with the auxiliary result columns dropped. +@dlt.table +def silver(): + df = dlt.read_stream("bronze_dq_check") + return dq_engine.get_valid(df) + +# COMMAND ---------- + +# Quarantine: only rows with errors or warnings. +@dlt.table +def quarantine(): + df = dlt.read_stream("bronze_dq_check") + return dq_engine.get_invalid(df) + +# COMMAND ---------- + +# Summary Metrics: materialized view computed by aggregation over the checked table. +# One row per metric (input / error / warning / valid row counts and per-check breakdown). +# Note: this MV is a cumulative snapshot over the whole table (input_row_count is the running +# total, not a per-run count). It refreshes incrementally only when the query is deterministic — +# set static run_time_overwrite / run_id_overwrite in ExtraParams for that. For per-run / per-window +# metrics on large or incrementally-checked tables, append from a windowed streaming table instead +# (see the "Snapshot vs. history" section of the Summary Metrics guide). +@dlt.table +def dq_summary_metrics(): + df = dlt.read("bronze_dq_check") + return dq_engine.compute_summary_metrics(df, checks=checks) diff --git a/docs/dqx/docs/demos.mdx b/docs/dqx/docs/demos.mdx index b65c27dcf..19802fe87 100644 --- a/docs/dqx/docs/demos.mdx +++ b/docs/dqx/docs/demos.mdx @@ -18,7 +18,8 @@ Import the following notebooks in the Databricks workspace to try DQX out: * [DQX Demo Notebook for Data Contract Integration (ODCS)](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_demo_datacontract_odcs.py) - demonstrates how to generate DQX quality rules from ODCS (Open Data Contract Standard) data contracts, including predefined rules from schema constraints, explicit custom rules, and contract metadata tracking. * [DQX Demo Notebook for Spark Structured Streaming (Native End-to-End Approach)](https://github.com/databrickslabs/dqx/blob/v0.15.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.15.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 using Spark API. -* [DQX Demo Notebook for Lakeflow Pipelines (formerly DLT)](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo.py) - demonstrates how to use DQX as a library with Lakeflow Pipelines. +* [DQX Demo Notebook for Lakeflow Pipelines (formerly DLT)](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo.py) - demonstrates how to use DQX in a Lakeflow Pipeline, reporting issues as additional columns and computing data quality summary metrics. +* [DQX Demo Notebook for Lakeflow Pipelines with Quarantine](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo_quarantine.py) - demonstrates how to use DQX in a Lakeflow Pipeline to split valid and invalid records into valid and quarantine tables and compute summary metrics. * [DQX usage with Declarative Automation Bundles](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_demo_asset_bundle/README.md) - demonstrates how to use DQX as a library with Declarative Automation Bundles (formerly Databricks Asset Bundles). * [DQX usage with dbt](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_demo_dbt/README.md) - demonstrates how to use DQX as a library with dbt projects. * [DQX Demo on Primary-Key detection](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_demo_llm_pk_detection.py) - demonstrates how to detect primary keys and generate uniqueness rules. diff --git a/docs/dqx/docs/guide/quality_checks_apply.mdx b/docs/dqx/docs/guide/quality_checks_apply.mdx index 3fab57d47..e4b3a53c5 100644 --- a/docs/dqx/docs/guide/quality_checks_apply.mdx +++ b/docs/dqx/docs/guide/quality_checks_apply.mdx @@ -1019,6 +1019,53 @@ To enable summary metrics programmatically, create and pass a `DQMetricsObserver +### Enabling summary metrics in Lakeflow Pipelines + +Inside a Lakeflow Pipeline (Spark Declarative Pipeline / DLT), do the following: + +1. Persist the checked data as a table (so the result columns are available downstream). +2. Add a materialized view that calls `compute_summary_metrics` on that table, passing the same `checks` you applied. + + + + ```python + import dlt + from databricks.labs.dqx.engine import DQEngine + from databricks.labs.dqx.metrics_observer import DQMetricsObserver + from databricks.sdk import WorkspaceClient + + # compute_summary_metrics requires an observer on the engine; it reads any custom_metrics from it. + dq_engine = DQEngine(WorkspaceClient(), observer=DQMetricsObserver()) + + @dlt.view + def bronze(): + return spark.readStream.table("catalog.schema.input") + + # 1. Apply checks and persist the checked data as a table. + @dlt.table + def silver(): + df = dlt.read_stream("bronze") + return dq_engine.apply_checks_by_metadata(df, checks) + + # 2. Compute summary metrics as a materialized view over the checked table. + @dlt.table + def dq_summary_metrics(): + df = dlt.read("silver") + return dq_engine.compute_summary_metrics(df, checks=checks) + ``` + + + +This works for both batch and streaming pipelines. See the [Lakeflow pipeline demo](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo.py) for a complete example, or the [quarantine variant](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo_quarantine.py) that splits valid and invalid records into separate tables. + + +The example is using a materialized view to compute the summary metrics. On each pipeline update the view is refreshed — **incrementally** when the query is deterministic (set static *run_time_overwrite* / *run_id_overwrite*, see the note above) and the engine can incrementalize the aggregation, otherwise fully recomputed. Either way it holds a **current snapshot** over all rows in the checked table, so `input_row_count` is the cumulative total, not a per-run count. + +To keep a **history** of metrics over time you need an append target: for **streaming** pipelines append per-window metrics from a windowed streaming table (recommended), and for **batch** pipelines append `compute_summary_metrics` output from a companion job. See the [Summary Metrics guide](/docs/guide/summary_metrics) for examples. + + +**Why not the observer directly?** The observer (`observe()`) and the streaming metrics listener do not work inside a declarative pipeline: the pipeline runtime — not your code — triggers the write action, so the Spark Observation is never populated (`observation.get` stalls) and the streaming listener receives no events. DQX detects the pipeline runtime and automatically skips wiring `observe()` there. `compute_summary_metrics` computes the same metrics as an aggregation instead, but still reads from the engine's observer — so **an observer must be configured** on the engine (calling it without one raises `InvalidParameterError`), and if you need **custom metrics**, configure it with them (`DQMetricsObserver(custom_metrics=[...])`). See the [Summary Metrics guide](/docs/guide/summary_metrics) for details. + ### Enabling summary metrics in DQX workflows Summary metrics can also be enabled in DQX workflows. Metrics are configured: diff --git a/docs/dqx/docs/guide/summary_metrics.mdx b/docs/dqx/docs/guide/summary_metrics.mdx index 7959a6346..df4ffd096 100644 --- a/docs/dqx/docs/guide/summary_metrics.mdx +++ b/docs/dqx/docs/guide/summary_metrics.mdx @@ -30,19 +30,22 @@ DQX automatically captures the following built-in metrics for every data quality When checks are applied, DQX automatically includes a `check_metrics` metric that provides a per-check breakdown of error and warning counts. This makes it possible to identify which specific checks are failing without querying the row-level `_errors` and `_warnings` columns. -The `check_metrics` value is a JSON-serialized array of structs, with one entry per check: +`check_metrics` contains an entry for **every applied check**, not only the ones that failed. Checks that never triggered still appear with `error_count` and `warning_count` set to `0`. This differs from the row-level `_errors` / `_warnings` columns, which list only the checks that failed for a given row. As a result, the breakdown is derived from the applied checks (their names), so it cannot be reconstructed from the data alone — a check with zero violations leaves no trace in `_errors` / `_warnings`. + +The `check_metrics` value is a JSON-serialized array of structs, with one entry per applied check (here `passenger_incorrect_count` was applied but never triggered): ```json [ {"check_name": "id_is_not_null", "error_count": 5, "warning_count": 0}, - {"check_name": "name_is_not_null_and_not_empty", "error_count": 0, "warning_count": 3} + {"check_name": "name_is_not_null_and_not_empty", "error_count": 0, "warning_count": 3}, + {"check_name": "passenger_incorrect_count", "error_count": 0, "warning_count": 0} ] ``` Each entry contains: - `check_name` — the name of the check (either explicitly set via `name` in the rule definition, or auto-derived from the check function and arguments) -- `error_count` — number of rows where this check triggered an error -- `warning_count` — number of rows where this check triggered a warning +- `error_count` — number of rows where this check triggered an error (`0` if it never did) +- `warning_count` — number of rows where this check triggered a warning (`0` if it never did) When persisted to the metrics table, `check_metrics` is stored as a single row with `metric_name = 'check_metrics'` and `metric_value` containing the JSON string. You can parse it with Spark's `from_json` or `json_tuple` functions for analysis: @@ -306,6 +309,85 @@ Metrics-only streaming writes are not supported because DQX needs an output or q +#### Summary Metrics in Spark Declarative Pipelines (Lakeflow / DLT) + +Inside a Spark Declarative Pipeline (SDP / Lakeflow / DLT), do the following instead of using an observer: + +1. Persist the checked data as a table (so the result columns are available downstream). +2. Add a materialized view that calls `compute_summary_metrics` on that table, passing the same `checks` you applied. + + + + ```python + import dlt + from databricks.labs.dqx.engine import DQEngine + from databricks.labs.dqx.metrics_observer import DQMetricsObserver + from databricks.sdk import WorkspaceClient + + # compute_summary_metrics requires an observer on the engine; it reads any custom_metrics from it. + engine = DQEngine(WorkspaceClient(), observer=DQMetricsObserver()) + + @dlt.view + def bronze(): + return spark.readStream.table("catalog.schema.input") + + # 1. Apply checks and persist the checked data as a table. + @dlt.table + def silver(): + df = dlt.read_stream("bronze") + return engine.apply_checks_by_metadata(df, checks) + + # 2. Compute summary metrics as a materialized view over the checked table. + @dlt.table + def dq_summary_metrics(): + df = dlt.read("silver") + return engine.compute_summary_metrics(df, checks=checks) + ``` + + + +This works whether the checked table is populated by a batch or a streaming pipeline — the metrics materialized view is recomputed (as batch) on each pipeline update. Note that `compute_summary_metrics` is a whole-table aggregation, not an append-mode streaming query; use it as a materialized view inside a pipeline (or in a batch job), not directly on a `writeStream` (a streaming aggregation without a watermark is not supported in append mode). See the [Lakeflow pipeline demo](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo.py) for a complete example, or the [quarantine variant](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo_quarantine.py) that splits valid and invalid records into separate tables. + + +`compute_summary_metrics` stamps each metrics row with a `run_id` and `run_time`. Without static values these are non-deterministic across pipeline runs — `run_time` defaults to `current_timestamp()` and `run_id` to a random UUID — so Lakeflow treats every update as new and cannot incrementally refresh the metrics materialized view (it fully recomputes each run). Set *run_time_overwrite* and *run_id_overwrite* as **static values** in `ExtraParams` when creating the `DQEngine` to enable incremental updates, the same as for [DQX result materialized views](/docs/guide/quality_checks_apply). + + + +The approach above stores summary metrics as a **materialized view** over the checked table. On each pipeline update the view is refreshed — **incrementally** when the query is deterministic (set static *run_time_overwrite* / *run_id_overwrite* as in the note above) and the pipeline engine can incrementalize the aggregation, otherwise fully recomputed. Either way it holds a **current snapshot**: it aggregates the whole table, so `input_row_count` is the cumulative total, not a per-run count. + +To instead keep a **history** of metrics over time (appended as the pipeline runs, so you can track quality trends), you need an append target rather than a materialized view. Choose based on the pipeline type: + +**Streaming pipelines (recommended):** append per-window metrics from a windowed streaming table. Add a watermark on a timestamp column and aggregate the observer's metric expressions per time window, so each window's metrics are appended as it completes: + +```python +import pyspark.sql.functions as F +from databricks.labs.dqx.metrics_observer import DQMetricsObserver + +observer = DQMetricsObserver() # pass check names to get_metrics(check_names) to also include the per-check breakdown + +@dlt.table +def dq_summary_metrics(): + df = dlt.read_stream("silver").withWatermark("event_time", "10 minutes") + metric_exprs = [F.expr(m) for m in observer.get_metrics()] + return df.groupBy(F.window("event_time", "1 hour")).agg(*metric_exprs) +``` + +This needs a timestamp column to window on (an event or ingestion time) and a watermark; each window's metrics are emitted only once the watermark passes it, and very late rows are dropped. The output carries a `window` column plus the metric columns (it is not the long `OBSERVATION_TABLE_SCHEMA` shape) — so it is **not** schema-compatible with the companion-job output below, and the two cannot be appended into the same metrics table. Pick one history strategy per table. + +**Batch pipelines (or a simpler option):** compute the metrics in a companion job or scheduled task and **append** `compute_summary_metrics` output — one row-set per run, stamped with a distinct `run_id` and `run_time`: + +```python +metrics_df = engine.compute_summary_metrics(spark.read.table("catalog.schema.silver"), checks=checks) +metrics_df.write.mode("append").saveAsTable("catalog.schema.dq_metrics") +``` + +**Multiple tables or pipelines:** a materialized view is defined by a single query in a single pipeline, so it cannot be shared. To centralize metrics from several checked tables in one pipeline, `unionByName` the `compute_summary_metrics` results inside one materialized view. To centralize across pipelines, use the companion-job pattern above to append into a plain Delta table (which accepts concurrent appends from multiple jobs); set a distinct observer `name` (and `input_config`) per source so rows stay filterable. + + +**Why not the observer directly?** The observer (`observe()`) and the streaming metrics listener do not work inside a declarative pipeline: the pipeline runtime — not your code — triggers the write action, so the Spark Observation is never populated (`observation.get` stalls) and the streaming listener receives no events. DQX detects the pipeline runtime and automatically skips wiring `observe()` there, so `apply_checks*` returns the checked DataFrame unchanged (no attached observation). `compute_summary_metrics` computes the same metrics as an aggregation instead, and returns a lazy DataFrame with the same schema as the observer path — so pass the same `checks` to get the full per-check breakdown (including checks with zero violations), and the metrics can be centralized alongside batch and streaming workloads. + +`compute_summary_metrics` still uses the engine's `DQMetricsObserver`, so **an observer must be configured** on the engine (`DQEngine(WorkspaceClient(), observer=DQMetricsObserver())`) — calling it without one raises `InvalidParameterError`. To include **custom metrics** in the pipeline, configure the observer with them (`DQMetricsObserver(custom_metrics=[...])`); `compute_summary_metrics` reads its `custom_metrics` and includes them in the output. Each custom metric expression must return a **scalar** value — a non-scalar result (e.g. an array or struct) is stringified opaquely into `metric_value`. See [Configuring Custom Metrics](#configuring-custom-metrics). + #### Saving Results and Metrics to a Table Summary metrics can also be written to a table when calling `save_results_in_table`. After applying checks, pass the Spark Observation and output DataFrame(s) with the appropriate output configuration. For batch results, you can pass only the Spark Observation and `metrics_config` to write summary metrics without writing row-level output. @@ -385,6 +467,8 @@ This is supported for both batch and streaming. Custom metrics are collected in addition to the built-in metrics. Pass custom metrics as Spark SQL expressions when creating the `DQMetricsObserver`. Custom metrics should be defined as Spark SQL expressions with column aliases and will be accessible by their alias. +Each custom metric expression must return a **scalar** aggregate value. `metric_value` is a string column, so a non-scalar result (e.g. an array or struct) is stringified opaquely and is hard to consume downstream — aggregate to a single scalar per metric. + ```python diff --git a/src/databricks/labs/dqx/engine.py b/src/databricks/labs/dqx/engine.py index f6e28952f..50a35cc57 100644 --- a/src/databricks/labs/dqx/engine.py +++ b/src/databricks/labs/dqx/engine.py @@ -51,7 +51,7 @@ from databricks.labs.dqx.metrics_observer import DQMetricsObservation, DQMetricsObserver from databricks.labs.dqx.metrics_listener import StreamingMetricsListener from databricks.labs.dqx.io import read_input_data, save_dataframe_as_table, get_reference_dataframes -from databricks.labs.dqx.telemetry import telemetry_logger, log_telemetry, log_dataframe_telemetry +from databricks.labs.dqx.telemetry import telemetry_logger, log_telemetry, log_dataframe_telemetry, is_dlt_pipeline from databricks.sdk import WorkspaceClient from databricks.labs.dqx.errors import InvalidCheckError, InvalidConfigError, InvalidParameterError from databricks.labs.dqx.utils import list_tables, safe_strip_file_from_path, resolve_variables, VariableValue @@ -291,7 +291,10 @@ def apply_checks_by_metadata_and_split( good_df, bad_df, *observations = self.apply_checks_and_split(df, dq_rule_checks, ref_dfs) - if self.observer: + # An observation is only returned when observe() was actually wired (an observer is set and we + # are not inside a Spark Declarative Pipeline, where it is skipped). Key off the returned shape + # rather than self.observer so the SDP path (observer set, observe() skipped) does not raise. + if observations: return good_df, bad_df, observations[0] return good_df, bad_df @@ -631,6 +634,19 @@ def _observe_metrics( if not self.observer: return df + # Inside a Spark Declarative Pipeline (SDP / Lakeflow / DLT) the runtime — not the caller — + # triggers the write, so an attached observe() never has an accessible result (observation.get + # stalls, the streaming listener receives no events). Skip wiring observe() there: apply_checks* + # returns the DataFrame unchanged (no tuple, no wasted/inaccessible observation), and metrics are + # instead computed by DQEngine.compute_summary_metrics over the checked table. The engine's + # observer (incl. its custom_metrics) is still used by that method. + if is_dlt_pipeline(self.spark): + logger.info( + "Spark Declarative Pipeline detected: observe()-based summary metrics are disabled. " + "Compute metrics with DQEngine.compute_summary_metrics(...) in a materialized view instead." + ) + return df + metric_exprs = [F.expr(m) for m in self.observer.get_metrics(check_names)] if not metric_exprs: return df @@ -1517,6 +1533,116 @@ def save_checks( handler = self._checks_handler_factory.create(config) handler.save(resolved_checks, config) + def _build_metrics_observation( + self, + observed_metrics: dict[str, Any] | None = None, + input_config: InputConfig | None = None, + output_config: OutputConfig | None = None, + quarantine_config: OutputConfig | None = None, + checks_location: str | None = None, + rule_set_fingerprint: str | None = None, + ) -> DQMetricsObservation: + """Build a *DQMetricsObservation* from the engine's run state and the given configs/metadata. + + Args: + observed_metrics: Collected summary metrics, when already available (the observe / streaming path). + input_config: Optional input configuration recorded for traceability. + output_config: Optional output configuration recorded for traceability. + quarantine_config: Optional quarantine configuration recorded for traceability. + checks_location: Optional checks location recorded for traceability. + rule_set_fingerprint: Optional SHA-256 fingerprint of the rule set used for this run. + + Returns: + A *DQMetricsObservation* populated from the engine's run id, run time, result column names, and metadata. + """ + return DQMetricsObservation( + run_id=self._engine.run_id, + run_name=self._engine.observer.name if self._engine.observer else DQMetricsObserver().name, + run_time_overwrite=self._engine.run_time_overwrite, + observed_metrics=observed_metrics, + error_column_name=self._engine.result_column_names[ColumnArguments.ERRORS], + warning_column_name=self._engine.result_column_names[ColumnArguments.WARNINGS], + input_location=input_config.location if input_config else None, + output_location=output_config.location if output_config else None, + quarantine_location=quarantine_config.location if quarantine_config else None, + checks_location=checks_location, + rule_set_fingerprint=rule_set_fingerprint, + user_metadata=self._engine.engine_user_metadata, + ) + + @telemetry_logger("engine", "compute_summary_metrics") + def compute_summary_metrics( + self, + checked_df: DataFrame, + checks: list[dict] | None = None, + custom_check_functions: dict[str, Callable] | None = None, + input_config: InputConfig | None = None, + output_config: OutputConfig | None = None, + quarantine_config: OutputConfig | None = None, + checks_location: str | None = None, + ) -> DataFrame: + """Compute data quality summary metrics from a checked DataFrame by aggregation. + + Unlike the observer/listener path (which relies on Spark *observe()* and a caller-triggered + action), this computes the same metrics as a plain aggregation over the result columns and + returns a lazy DataFrame. This makes it usable inside Spark Declarative Pipelines (SDP / + Lakeflow / DLT), where the pipeline runtime — not the caller — owns the write action: define a + downstream materialized view over the checked table that returns the result of this method. + + Args: + checked_df: DataFrame produced by *apply_checks* / *apply_checks_by_metadata* (must still + contain the DQX result columns, i.e. before *get_valid* / *get_invalid* drop them). + checks: Optional metadata checks that were applied (the same list of dicts passed to + *apply_checks_by_metadata*). When provided, a per-check breakdown (*check_metrics*) is + included covering every applied check, including checks with zero violations. The breakdown + is derived from the check names and cannot be reconstructed from data alone, so pass the + same checks used when applying. When omitted, only dataset-level metrics (row counts and + any observer custom metrics) are produced. + custom_check_functions: Optional custom check functions used to resolve metadata checks. Pass the + *same* functions that were used when the checks were applied — if the applied checks referenced a + custom function and it is not supplied here, deserialization fails or resolves a different check + name, so the *check_metrics* breakdown and *rule_set_fingerprint* would not match the applied run. + input_config: Optional input configuration recorded in the metrics for traceability. + output_config: Optional output configuration recorded in the metrics for traceability. + quarantine_config: Optional quarantine configuration recorded in the metrics for traceability. + checks_location: Optional checks location recorded in the metrics for traceability. + + Note: + A *DQMetricsObserver* must be configured on this engine (*DQEngine(..., observer=...)*); its + *custom_metrics* (if any) are included alongside the built-in dataset-level metrics and the + per-check breakdown (when *checks* is provided). + + Returns: + A lazy DataFrame matching *OBSERVATION_TABLE_SCHEMA* with one row per metric. + + Raises: + InvalidParameterError: If no *DQMetricsObserver* is configured on the engine. + """ + observer = self._engine.observer + if observer is None: + raise InvalidParameterError( + "Summary metrics cannot be computed for an engine with no observer. " + "Configure a DQMetricsObserver on the engine, e.g. DQEngine(workspace_client, observer=DQMetricsObserver(...))." + ) + + check_names: list[str] | None = None + rule_set_fingerprint: str | None = None + if checks: + rules = deserialize_checks(checks, custom_check_functions) + # Duplicate check names are preserved so check_metrics reports each occurrence separately. + check_names = [rule.name for rule in rules] + rule_set_fingerprint = compute_rule_set_fingerprint(rules) + + aggregated_df = checked_df.selectExpr(*observer.get_metrics(check_names)) + observation = self._build_metrics_observation( + input_config=input_config, + output_config=output_config, + quarantine_config=quarantine_config, + checks_location=checks_location, + rule_set_fingerprint=rule_set_fingerprint, + ) + return DQMetricsObserver.build_metrics_df_from_aggregation(aggregated_df, observation) + @telemetry_logger("engine", "save_summary_metrics") def save_summary_metrics( self, @@ -1550,20 +1676,13 @@ def save_summary_metrics( This method is only supported by spark batch. Spark query listener must be used for streaming: For streaming use spark.streams.addListener(get_streaming_metrics_listener(..)) """ - run_name = self._engine.observer.name if self._engine.observer else DQMetricsObserver().name - metrics_observation = DQMetricsObservation( - run_id=self._engine.run_id, - run_name=run_name, - run_time_overwrite=self._engine.run_time_overwrite, + metrics_observation = self._build_metrics_observation( observed_metrics=observed_metrics, - error_column_name=self._engine.result_column_names[ColumnArguments.ERRORS], - warning_column_name=self._engine.result_column_names[ColumnArguments.WARNINGS], - input_location=input_config.location if input_config else None, - output_location=output_config.location if output_config else None, - quarantine_location=quarantine_config.location if quarantine_config else None, + input_config=input_config, + output_config=output_config, + quarantine_config=quarantine_config, checks_location=checks_location, rule_set_fingerprint=rule_set_fingerprint, - user_metadata=self._engine.engine_user_metadata, ) metrics_df = DQMetricsObserver.build_metrics_df(self.spark, metrics_observation) @@ -1607,18 +1726,12 @@ def get_streaming_metrics_listener( self._validate_metrics_observer(metrics_config) assert self._engine.observer is not None # guaranteed by _validate_metrics_observer above (required by mypy) - metrics_observation = DQMetricsObservation( - run_id=self._engine.run_id, - run_name=self._engine.observer.name, - run_time_overwrite=self._engine.run_time_overwrite, - error_column_name=self._engine.result_column_names[ColumnArguments.ERRORS], - warning_column_name=self._engine.result_column_names[ColumnArguments.WARNINGS], - input_location=input_config.location if input_config else None, - output_location=output_config.location if output_config else None, - quarantine_location=quarantine_config.location if quarantine_config else None, + metrics_observation = self._build_metrics_observation( + input_config=input_config, + output_config=output_config, + quarantine_config=quarantine_config, checks_location=checks_location, rule_set_fingerprint=rule_set_fingerprint, - user_metadata=self._engine.engine_user_metadata, ) return StreamingMetricsListener(metrics_config, metrics_observation, self.spark, target_query_id) diff --git a/src/databricks/labs/dqx/metrics_observer.py b/src/databricks/labs/dqx/metrics_observer.py index 756fed8bd..f87187849 100644 --- a/src/databricks/labs/dqx/metrics_observer.py +++ b/src/databricks/labs/dqx/metrics_observer.py @@ -211,3 +211,77 @@ def build_metrics_df(spark: SparkSession, observation: DQMetricsObservation) -> df = df.withColumn("run_time", F.current_timestamp()) return df + + @staticmethod + def build_metrics_df_from_aggregation(aggregated_df: DataFrame, observation: DQMetricsObservation) -> DataFrame: + """ + Reshapes a one-row wide aggregation of metric expressions into the long-format + *OBSERVATION_TABLE_SCHEMA*, without triggering a Spark action. + + Used by *DQEngine.compute_summary_metrics* to keep a lazily-computed aggregation lazy, so the + result can back a materialized view or table in a Spark Declarative Pipeline (where the pipeline + runtime — not the caller — triggers the write). The already-collected path uses *build_metrics_df*. + + The input must be a **single-row global aggregation** (the output of *get_metrics* selected with no + *groupBy*). A multi-row input would emit one metrics row-set per input row, each stamped with the + same run metadata and no grouping key, so this must not be used for windowed/grouped aggregations. + + Args: + aggregated_df: A single-row DataFrame whose columns are the metric expressions produced by + *DQMetricsObserver.get_metrics* (e.g. *input_row_count*, *error_row_count*, *check_metrics*). + observation: *DQMetricsObservation* carrying the run metadata (locations, fingerprint, run time). + + Returns: + A lazy Spark DataFrame matching *OBSERVATION_TABLE_SCHEMA* with one row per metric. + """ + # Unpivot the wide one-row aggregation into (metric_name, metric_value) rows using the DataFrame + # API: build one struct per metric (name literal + value cast to string) and explode. + # + # Reference the metric columns by safe positional names, not by their real names: a metric name + # can contain a dot (e.g. a user-supplied custom_metrics alias), and `aggregated_df["a.b"]` would + # parse the dot as nested-field access ("field b of column a") rather than a top-level column, + # misresolving or raising. `toDF` renames the columns positionally (without resolving the old + # names), so the value lookup uses a dot-free name; the real name is only used as a literal for + # metric_name, where a dot is harmless. This avoids SQL-string building and identifier escaping. + metric_names = aggregated_df.columns + renamed = aggregated_df.toDF(*[f"metric_{i}" for i in range(len(metric_names))]) + metric_pairs = F.array( + *[ + F.struct( + F.lit(metric_names[i]).alias("metric_name"), + renamed[f"metric_{i}"].cast("string").alias("metric_value"), + ) + for i in range(len(metric_names)) + ] + ) + long_df = renamed.select(F.explode(metric_pairs).alias("metric")) + + # run_time_overwrite is a datetime, so F.lit produces a timestamp literal via the same + # Python->Spark conversion createDataFrame uses in build_metrics_df — run_time is therefore + # identical across both builders. Fall back to current_timestamp() when it is not set. + if observation.run_time_overwrite is not None: + run_time = F.lit(observation.run_time_overwrite).cast("timestamp") + else: + run_time = F.current_timestamp() + + if observation.user_metadata: + map_entries = [F.lit(item) for pair in observation.user_metadata.items() for item in pair] + user_metadata = F.create_map(*map_entries) + else: + user_metadata = F.lit(None).cast("map") + + return long_df.select( + F.lit(observation.run_id).cast("string").alias("run_id"), + F.lit(observation.run_name).cast("string").alias("run_name"), + F.lit(observation.input_location).cast("string").alias("input_location"), + F.lit(observation.output_location).cast("string").alias("output_location"), + F.lit(observation.quarantine_location).cast("string").alias("quarantine_location"), + F.lit(observation.checks_location).cast("string").alias("checks_location"), + F.lit(observation.rule_set_fingerprint).cast("string").alias("rule_set_fingerprint"), + F.col("metric.metric_name").alias("metric_name"), + F.col("metric.metric_value").alias("metric_value"), + run_time.alias("run_time"), + F.lit(observation.error_column_name).cast("string").alias("error_column_name"), + F.lit(observation.warning_column_name).cast("string").alias("warning_column_name"), + user_metadata.alias("user_metadata"), + ) diff --git a/tests/e2e/test_run_demos.py b/tests/e2e/test_run_demos.py index 0a0e3ffee..717302b85 100644 --- a/tests/e2e/test_run_demos.py +++ b/tests/e2e/test_run_demos.py @@ -8,6 +8,7 @@ from uuid import uuid4 from tempfile import TemporaryDirectory +import pytest import yaml from databricks.sdk.service.workspace import ImportFormat @@ -186,17 +187,18 @@ def test_run_dqx_demo_pii_detection(ws, make_notebook, make_job, library_ref): logging.info(f"Job run {run.run_id} completed successfully for dqx_demo_pii_detection") +@pytest.mark.parametrize("demo_notebook", ["dqx_dlt_demo.py", "dqx_dlt_demo_quarantine.py"]) def test_run_dqx_dlt_demo( - skip_if_classic_compute, ws, make_notebook, make_schema, make_pipeline, make_job, library_ref + skip_if_classic_compute, ws, make_notebook, make_schema, make_pipeline, make_job, library_ref, demo_notebook ): """ - Test running the DLT demo notebook in a serverless pipeline. + Test running the DLT demo notebooks (report-as-columns and quarantine variants) in a serverless pipeline. No need to trigger from non-serverless runtime, since the dlt pipeline use own cluster anyway. """ catalog = TEST_CATALOG schema = make_schema(catalog_name=catalog).name - path = Path(__file__).parent.parent.parent / "demos" / "dqx_dlt_demo.py" + path = Path(__file__).parent.parent.parent / "demos" / demo_notebook with open(path, "rb") as f: notebook = make_notebook(content=f, format=ImportFormat.SOURCE) @@ -215,7 +217,8 @@ def test_run_dqx_dlt_demo( environment=PipelinesEnvironment(dependencies=[library_ref]), ) pipeline_task = PipelineTask(pipeline_id=pipeline.pipeline_id) - job = make_job(tasks=[Task(task_key="dqx_dlt_demo", pipeline_task=pipeline_task)]) + task_key = demo_notebook.removesuffix(".py") + job = make_job(tasks=[Task(task_key=task_key, pipeline_task=pipeline_task)]) waiter = ws.jobs.run_now_and_wait(job.job_id, timeout=timedelta(minutes=30)) run = ws.jobs.wait_get_run_job_terminated_or_skipped( @@ -223,7 +226,7 @@ def test_run_dqx_dlt_demo( timeout=timedelta(minutes=30), callback=lambda r: validate_run_status(r, ws), ) - logging.info(f"Job run {run.run_id} completed successfully for dqx_dlt_demo") + logging.info(f"Job run {run.run_id} completed successfully for {demo_notebook}") def test_run_dqx_demo_tool(ws, installation_ctx, make_schema, make_notebook, make_job): @@ -515,11 +518,12 @@ def test_run_dqx_row_anomaly_detection_demo(ws, make_notebook, make_schema, make job = make_job(tasks=[Task(task_key="dqx_row_anomaly_detection_demo", notebook_task=notebook_task)]) # This demo trains two IsolationForest models and scores with contributions + AI explanations - # (on by default), so it runs past run_now_and_wait's 20-minute SDK default - waiter = ws.jobs.run_now_and_wait(job.job_id, timeout=timedelta(minutes=30)) + # (on by default), so it is the slowest e2e demo and can exceed 30 minutes on a cold serverless + # start. Use a 45-minute wait (still well within the e2e CI job's 2h wrapper). + waiter = ws.jobs.run_now_and_wait(job.job_id, timeout=timedelta(minutes=45)) run = ws.jobs.wait_get_run_job_terminated_or_skipped( run_id=waiter.run_id, - timeout=timedelta(minutes=30), + timeout=timedelta(minutes=45), callback=lambda r: validate_run_status(r, ws), ) logging.info(f"Job run {run.run_id} completed successfully for dqx_row_anomaly_detection_demo") diff --git a/tests/integration/test_summary_metrics.py b/tests/integration/test_summary_metrics.py index 38f2a1437..9168da642 100644 --- a/tests/integration/test_summary_metrics.py +++ b/tests/integration/test_summary_metrics.py @@ -21,7 +21,7 @@ from databricks.labs.dqx.reporting_columns import ColumnArguments from tests.constants import TEST_CATALOG -from tests.integration.conftest import EXTRA_PARAMS +from tests.integration.conftest import EXTRA_PARAMS, RUN_TIME # Test constants TEST_SCHEMA = StructType( @@ -3092,3 +3092,144 @@ def _assert_check_metrics(actual_metrics: dict, expected_check_names: list[str]) check_metrics = json.loads(actual_metrics["check_metrics"]) actual_names = [m["check_name"] for m in check_metrics] assert actual_names == expected_check_names + + +def _standard_test_df(spark): + """Standard 4-row test frame: row 3 (id=None) errors, row 4 (name=None) warns under TEST_CHECKS.""" + return spark.createDataFrame( + [ + [1, "Alice", 30, 50000], + [2, "Bob", 25, 45000], + [None, "Charlie", 35, 60000], + [4, None, 28, 55000], + ], + TEST_SCHEMA, + ) + + +def test_compute_summary_metrics(ws, spark): + """compute_summary_metrics produces the full metrics set by aggregation, without observe() or an action.""" + observer = DQMetricsObserver(name=TEST_OBSERVER_NAME) + dq_engine = DQEngine(workspace_client=ws, spark=spark, observer=observer, extra_params=EXTRA_PARAMS) + + checked_df, _ = dq_engine.apply_checks_by_metadata(_standard_test_df(spark), TEST_CHECKS) + + input_config = InputConfig(location="input_table") + output_config = OutputConfig(location="output_table") + quarantine_config = OutputConfig(location="quarantine_table") + checks_location = "checks_location" + + metrics_df = dq_engine.compute_summary_metrics( + checked_df, + checks=TEST_CHECKS, + input_config=input_config, + output_config=output_config, + quarantine_config=quarantine_config, + checks_location=checks_location, + ).orderBy("metric_name") + + def _row(metric_name: str, metric_value: str) -> dict: + return { + "run_id": EXTRA_PARAMS.run_id_overwrite, + "run_name": TEST_OBSERVER_NAME, + "input_location": input_config.location, + "output_location": output_config.location, + "quarantine_location": quarantine_config.location, + "checks_location": checks_location, + "rule_set_fingerprint": TEST_CHECKS_RULE_SET_FINGERPRINT, + "metric_name": metric_name, + "metric_value": metric_value, + "run_time": RUN_TIME, + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + } + + expected_metrics = [ + _row("check_metrics", TEST_CHECK_METRICS_VALUE), + _row("error_row_count", "1"), + _row("input_row_count", "4"), + _row("valid_row_count", "2"), + _row("warning_row_count", "1"), + ] + expected_metrics_df = spark.createDataFrame(expected_metrics, schema=OBSERVATION_TABLE_SCHEMA).orderBy( + "metric_name" + ) + + assertDataFrameEqual(expected_metrics_df, metrics_df) + + +@pytest.mark.parametrize("apply_checks_method", [DQEngine.apply_checks, DQEngine.apply_checks_by_metadata]) +def test_compute_summary_metrics_matches_observer(ws, spark, apply_checks_method): + """Aggregation-based metrics match the observe()-based metrics for the same data and checks (parity).""" + observer = DQMetricsObserver(name="test_observer") + dq_engine = DQEngine(workspace_client=ws, spark=spark, observer=observer, extra_params=EXTRA_PARAMS) + + test_df = _standard_test_df(spark) + + if apply_checks_method == DQEngine.apply_checks: + checked_df, observation = dq_engine.apply_checks(test_df, deserialize_checks(TEST_CHECKS)) + else: + checked_df, observation = dq_engine.apply_checks_by_metadata(test_df, TEST_CHECKS) + + checked_df.count() # trigger the action so the observe()-based metrics populate + observed_metrics = observation.get + + # compute_summary_metrics takes metadata checks; TEST_CHECKS yields the same names/breakdown. + metrics_df = dq_engine.compute_summary_metrics(checked_df, checks=TEST_CHECKS) + computed_metrics = {row["metric_name"]: row["metric_value"] for row in metrics_df.collect()} + + # observe() returns native types (ints); compute_summary_metrics returns the long-format string values. + expected_metrics = {name: str(value) for name, value in observed_metrics.items()} + assert computed_metrics == expected_metrics + + +def test_compute_summary_metrics_without_checks(ws, spark): + """Without checks only the dataset-level metrics are produced; the per-check breakdown is omitted.""" + observer = DQMetricsObserver(name=TEST_OBSERVER_NAME) + dq_engine = DQEngine(workspace_client=ws, spark=spark, observer=observer, extra_params=EXTRA_PARAMS) + + checked_df, _ = dq_engine.apply_checks_by_metadata(_standard_test_df(spark), TEST_CHECKS) + + metrics_df = dq_engine.compute_summary_metrics(checked_df) + metrics = {row["metric_name"]: row["metric_value"] for row in metrics_df.collect()} + + assert metrics == { + "input_row_count": "4", + "error_row_count": "1", + "warning_row_count": "1", + "valid_row_count": "2", + } + + +def test_compute_summary_metrics_with_dotted_custom_metric_name(ws, spark): + """A custom metric aliased with a dotted name must be emitted as-is, not misread as nested-field access.""" + observer = DQMetricsObserver(name="test_observer", custom_metrics=["count(1) as `a.b`"]) + dq_engine = DQEngine(workspace_client=ws, spark=spark, observer=observer, extra_params=EXTRA_PARAMS) + + checked_df, _ = dq_engine.apply_checks_by_metadata(_standard_test_df(spark), TEST_CHECKS) + + metrics_df = dq_engine.compute_summary_metrics(checked_df, checks=TEST_CHECKS) + metrics = {row["metric_name"]: row["metric_value"] for row in metrics_df.collect()} + + # The dotted alias flows through as the metric name with its aggregated value (count over 4 rows). + assert metrics["a.b"] == "4" + + +def test_compute_summary_metrics_current_timestamp_and_user_metadata(ws, spark): + """Without run_time_overwrite the run_time falls back to current_timestamp(); user_metadata is emitted as a map.""" + user_metadata = {"team": "data-quality", "env": "test"} + observer = DQMetricsObserver(name=TEST_OBSERVER_NAME) + # No run_time_overwrite, so build_metrics_df_from_aggregation stamps run_time with current_timestamp(). + extra_params = ExtraParams(user_metadata=user_metadata) + dq_engine = DQEngine(workspace_client=ws, spark=spark, observer=observer, extra_params=extra_params) + + checked_df, _ = dq_engine.apply_checks_by_metadata(_standard_test_df(spark), TEST_CHECKS) + + rows = dq_engine.compute_summary_metrics(checked_df, checks=TEST_CHECKS).collect() + + assert rows # metrics were produced + # run_time falls back to current_timestamp() (non-null) when no run_time_overwrite is configured. + assert all(row["run_time"] is not None for row in rows) + # user_metadata is emitted as a map on every metric row. + assert all(row["user_metadata"] == user_metadata for row in rows) diff --git a/tests/unit/test_engine.py b/tests/unit/test_engine.py index 6e15829ec..96f4074c2 100644 --- a/tests/unit/test_engine.py +++ b/tests/unit/test_engine.py @@ -529,3 +529,54 @@ def test_apply_checks_parses_dbr_version_with_suffix(dbr_version, required_check match=rf"require Databricks Runtime >= .*but the current version is {re.escape(dbr_version)}", ): engine.apply_checks(df, [DQRowRule(check_func=required_check, column="a")]) + + +def _observer_core_with_pipeline_conf(pipelines_id: str | None) -> DQEngineCore: + """Build a DQEngineCore with an observer and a spark whose 'pipelines.id' conf is (un)set.""" + spark_mock = create_autospec(SparkSession) + spark_mock.conf = Mock() + spark_mock.conf.get.return_value = pipelines_id # None => not a pipeline; a value => inside a pipeline + ws = create_autospec(WorkspaceClient) + observer = DQMetricsObserver(name="test_observer") + return DQEngineCore(spark=spark_mock, workspace_client=ws, observer=observer) + + +def test_apply_checks_skips_observe_in_declarative_pipeline(): + """Inside a Spark Declarative Pipeline, apply_checks returns a bare DataFrame and does not wire observe().""" + core = _observer_core_with_pipeline_conf("pipeline-123") + + checked_df = Mock() + checked_df.observe.return_value = Mock() + df = Mock() + df.columns = ["id", "name"] + df.select.return_value = checked_df # _append_empty_checks builds the checked df via df.select(...) + + result = core.apply_checks(df, []) + + # observe() is inaccessible in a pipeline, so it is skipped: bare DataFrame, no observation, no observe() call. + assert not isinstance(result, tuple) + checked_df.observe.assert_not_called() + + +def test_apply_checks_wires_observe_outside_declarative_pipeline(): + """Outside a Spark Declarative Pipeline, apply_checks wires observe() and returns a (DataFrame, Observation).""" + core = _observer_core_with_pipeline_conf(None) + + checked_df = Mock() + checked_df.isStreaming = False + checked_df.observe.return_value = Mock() + df = Mock() + df.columns = ["id", "name"] + df.select.return_value = checked_df + + result = core.apply_checks(df, []) + + assert isinstance(result, tuple) + checked_df.observe.assert_called_once() + + +def test_compute_summary_metrics_raises_without_observer(mock_workspace_client, mock_spark): + """compute_summary_metrics requires an observer on the engine and raises when none is configured.""" + engine = DQEngine(mock_workspace_client, mock_spark) + with pytest.raises(InvalidParameterError, match="no observer"): + engine.compute_summary_metrics(Mock(), checks=[])