Add summary metrics #553
Open
Add summary metrics #553
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
❌ 388/404 passed, 5 flaky, 16 failed, 18 skipped, 4h47m15s total ❌ test_e2e_workflow_with_custom_install_folder: [gw5] linux -- Python 3.12.11 /home/runner/work/dqx/dqx/.venv/bin/python (726ms)❌ test_quality_checker_workflow_with_metrics: pyspark.errors.exceptions.connect.AnalysisException: [TABLE_OR_VIEW_NOT_FOUND] The table or view `main`.`dummy_scowo`.`metrics_gttua7` cannot be found. Verify the spelling and correctness of the schema and catalog. (2m40.658s)❌ test_custom_metrics_in_workflow: pyspark.errors.exceptions.connect.AnalysisException: [TABLE_OR_VIEW_NOT_FOUND] The table or view `main`.`dummy_sccc2`.`metrics_pm3txo` cannot be found. Verify the spelling and correctness of the schema and catalog. (3m1.997s)❌ test_quality_checker_workflow_with_quarantine_and_metrics: databricks.sdk.errors.platform.Unknown: apply_checks: Run failed with error message (1m5.863s)❌ test_profiler_workflow_with_custom_install_folder: [gw9] linux -- Python 3.12.11 /home/runner/work/dqx/dqx/.venv/bin/python (812ms)❌ test_e2e_workflow_for_multiple_run_configs: databricks.sdk.errors.platform.Unknown: prepare: Run failed with error message (1m33.08s)❌ test_quality_checker_workflow_for_multiple_run_configs: databricks.labs.dqx.errors.InvalidConfigError: Run config flag is required (2m47.981s)❌ test_e2e_workflow_with_metrics: pyspark.errors.exceptions.connect.AnalysisException: [TABLE_OR_VIEW_NOT_FOUND] The table or view `main`.`dummy_selzg`.`metrics_twfapr` cannot be found. Verify the spelling and correctness of the schema and catalog. (8m28.732s)❌ test_quality_checker_workflow_for_multiple_run_configs_table_checks_storage: databricks.labs.dqx.errors.InvalidConfigError: Run config flag is required (3m11.575s)❌ test_observer_metrics_output: chispa.dataframe_comparer.DataFramesNotEqualError: (11.486s)❌ test_observer_metrics_output_with_quarantine: chispa.dataframe_comparer.DataFramesNotEqualError: (13.597s)❌ test_streaming_observer_metrics_output: pyspark.errors.exceptions.connect.AnalysisException: [TABLE_OR_VIEW_NOT_FOUND] The table or view `main`.`dummy_seqiy`.`metrics_p3llvn` cannot be found. Verify the spelling and correctness of the schema and catalog. (12.845s)❌ test_save_results_in_table_batch_with_metrics: chispa.dataframe_comparer.DataFramesNotEqualError: (10.446s)❌ test_save_results_in_table_streaming_with_metrics: pyspark.errors.exceptions.connect.AnalysisException: [TABLE_OR_VIEW_NOT_FOUND] The table or view `main`.`dummy_sj69h`.`metrics_bqkexy` cannot be found. Verify the spelling and correctness of the schema and catalog. (11.513s)❌ test_quality_checker_workflow_with_custom_install_folder: [gw6] linux -- Python 3.12.11 /home/runner/work/dqx/dqx/.venv/bin/python (630ms)❌ test_list_tables: databricks.sdk.errors.platform.NotFound: Catalog 'ucx_9tmnasv5gbmchcct' does not exist. (18m39.399s)Flaky tests:
Running from acceptance #2724 |
There was a problem hiding this comment.
Pull Request Overview
This PR introduces summary metrics as outputs of quality checking methods, using Spark's Observation feature to track data quality metrics. The DQObserver class manages Spark observations and tracks both default metrics (input/error/warning/valid counts) and custom user-defined SQL expressions.
- Adds
DQObserverclass for managing Spark observations and tracking summary metrics - Updates
DQEnginemethods to return tuples with both DataFrames and observations - Integrates metrics collection with existing workflows and configuration system
Reviewed Changes
Copilot reviewed 17 out of 17 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| src/databricks/labs/dqx/observer.py | New DQObserver class for managing Spark observations and tracking metrics |
| src/databricks/labs/dqx/engine.py | Updated engine methods to support metrics collection and storage |
| src/databricks/labs/dqx/config.py | Added metrics configuration fields to RunConfig and WorkspaceConfig |
| tests/unit/test_observer.py | Unit tests for the DQObserver class functionality |
| tests/integration/test_summary_metrics.py | Integration tests for end-to-end metrics collection |
| tests/integration/test_metrics_workflow.py | Tests for workflow-based metrics collection |
Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.
src/databricks/labs/dqx/engine.py
Outdated
Comment on lines
621
to
622
| save_dataframe_as_table(metrics_df, metrics_config) | ||
| save_dataframe_as_table(metrics_df, metrics_config) |
| assert run_config.metrics_config is None | ||
|
|
||
| ctx.deployed_workflows.run_workflow("quality-checker", run_config.name) | ||
| assert not ws.tables.exists(run_config.metrics_config.location).table_exists |
There was a problem hiding this comment.
This assertion will fail because run_config.metrics_config is None when metrics are disabled, causing a NoneType attribute access error. The assertion should check that run_config.metrics_config is None instead.
Suggested change
| assert not ws.tables.exists(run_config.metrics_config.location).table_exists | |
| # Cannot check for metrics table existence as metrics_config is None. |
Copilot uses AI. Check for mistakes.
mwojtyczka
requested changes
Aug 29, 2025
| - `sample_seed`: seed for reproducible sampling. | ||
| - `limit`: maximum number of records to analyze. | ||
| - `extra_params`: (optional) extra parameters to pass to the jobs such as result column names and user_metadata | ||
| - `custom_metrics`: (optional) list of Spark SQL expressions for capturing custom summary metrics. |
There was a problem hiding this comment.
it would be worth to add that a set of default metrics are always used regardless.
There was a problem hiding this comment.
Do you think this is clear?
By default, the number of input, warning, and error rows will be tracked. When custom metrics are defined, they will be tracked in addition to the default metrics.
src/databricks/labs/dqx/observer.py
Outdated
| A list of Spark SQL expressions as strings | ||
| """ | ||
| result_columns = self.result_columns or {} | ||
| errors_column = result_columns.get(ColumnArguments.ERRORS.value, DefaultColumnNames.ERRORS.value) |
There was a problem hiding this comment.
it would be better to provide this directly to avoid repeating the engine implementation. I would set this in the engine. Otherwise, user have to provide extra params twice and it would be error prone.
There was a problem hiding this comment.
We need a way to update the column names in SQL expressions used for default metrics. I added a _set_column_names method in the DQMetricsObserver that is called whenever DQEngine is initialized with an observer.
src/databricks/labs/dqx/observer.py
Outdated
| """ | ||
| Spark `Observation` which can be attached to a `DataFrame` to track summary metrics. Metrics will be collected | ||
| when the 1st action is triggered on the attached `DataFrame`. Subsequent operations on the attached `DataFrame` | ||
| will not update the observed metrics. See: https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.Observation.html |
There was a problem hiding this comment.
make this a proper link so that we render this nicely in api docs
| assert run_config.metrics_config is None | ||
|
|
||
| ctx.deployed_workflows.run_workflow("quality-checker", run_config.name) | ||
| assert not ws.tables.exists(run_config.metrics_config.location).table_exists |
|
|
||
| def test_engine_with_observer_before_action(ws, spark): | ||
| """Test that summary metrics are empty before running a Spark action.""" | ||
| custom_metrics = ["avg(age) as avg_age", "sum(salary) as total_salary"] |
There was a problem hiding this comment.
do we need custom metrics here? it seems we can remove it for this test
updated docs
…trics # Conflicts: # docs/dqx/docs/reference/benchmarks.mdx # src/databricks/labs/dqx/engine.py # tests/perf/.benchmarks/baseline.json
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Changes
This PR introduces summary metrics as outputs of quality checking methods. Summary metrics computation relies on Spark's Observation feature.
Basic usage
The
DQObservercan be added toDQEngineto manage Spark observations and track summary metrics on datasets checked with DQX:Methods of
DQEnginehave been updated to optionally return the Spark observation associated with a given run:Writing summary metrics with checked data
When
DQEnginemethods write results to an output sink, metrics can also be written:Integration with installed workflows
I have also updated the quality checking and e2e workflows to allow users to specify an output table where metrics are stored.
TODO
Linked issues
Resolves #376
Tests