diff --git a/docs/dqx/docs/guide/quality_checks_apply.mdx b/docs/dqx/docs/guide/quality_checks_apply.mdx index 198a39f56..b98ec403c 100644 --- a/docs/dqx/docs/guide/quality_checks_apply.mdx +++ b/docs/dqx/docs/guide/quality_checks_apply.mdx @@ -14,6 +14,7 @@ You can apply quality checks to your data using the following approaches: * [No-code approach (Workflows)](#no-code-approach-workflows): you can apply checks for data at-rest using workflows if DQX is installed in the workspace as a tool. This does not require code-level integration and is suitable for data already stored in Delta tables or files. DQX offers a set of predefined built-in quality rules (checks) as described [here](/docs/guide/quality_checks_definition). +All DQX rules are evaluated independently and in parallel. If one rule fails, the others are still processed. ## Programmatic approach @@ -30,7 +31,8 @@ Checks can be applied to the input data by one of the following methods of the ` You can see the full list of `DQEngine` methods [here](/docs/reference/engine/#dqx-engine-methods). -The engine will raise an error if you try to apply checks with invalid definition. +The engine ensures that the specified `column`, `columns`, or `filter` fields can be resolved in the input DataFrame. If any of these fields are invalid, the check evaluation is skipped, and the results include the check failure with a message identifying the invalid fields. +The engine will raise an error if you try to apply checks with invalid definition (e.g. wrong syntax). In addition, you can also perform a standalone syntax validation of the checks as described [here](/docs/guide/quality_checks_definition#validating-syntax-of-quality-checks). You can apply quality checks to streaming pipelines using the same methods as for batch processing. diff --git a/docs/dqx/docs/guide/quality_checks_definition.mdx b/docs/dqx/docs/guide/quality_checks_definition.mdx index b742d03a2..ecb94be73 100644 --- a/docs/dqx/docs/guide/quality_checks_definition.mdx +++ b/docs/dqx/docs/guide/quality_checks_definition.mdx @@ -553,17 +553,17 @@ Checks defined in delta tables are supported by the DQX workflows. The Delta table used to store checks has the following structure: -| Column | Type | Description | -| -------------------- | --------------------- | ---------------------------------------------------------------------------------------------------------------------------------------- | -| `name` | `string` | Name to use for the check. Optional. | -| `criticality` | `string` | Either `"error"` (rows go only to "bad" dataset) or `"warn"` (rows go to both "good" and "bad"). Defaults to `"error"`. | -| `check` | `struct` | Defines the DQX check to apply. | -| └─ `function` | `string` | Name of the DQX check function to apply. | -| └─ `arguments` | `map` | Key-value pairs passed as keyword arguments to the function, stored as JSON string so quotes for string arguments must be escaped. | -| └─ `for_each_column` | `array` | (Optional) List of column names or expressions to which the check will be applied individually. | +| Column | Type | Description | +| -------------------- | --------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `name` | `string` | Name to use for the check. Optional. | +| `criticality` | `string` | Either `"error"` (rows go only to "bad" dataset) or `"warn"` (rows go to both "good" and "bad"). Defaults to `"error"`. | +| `check` | `struct` | Defines the DQX check to apply. | +| └─ `function` | `string` | Name of the DQX check function to apply. | +| └─ `arguments` | `map` | Key-value pairs passed as keyword arguments to the function, stored as JSON string so quotes for string arguments must be escaped. | +| └─ `for_each_column` | `array` | (Optional) List of column names or expressions to which the check will be applied individually. | | `filter` | `string` | (Optional) Spark SQL expression to filter rows to which the check is applied, e.g. `"business_unit = 'Finance'"`. The check function will run only on the rows matching the filter condition. The condition can reference any column of the validated dataset, not only the one where you apply the check function. | -| `run_config_name` | `string` | Name of the run config name. Could be any string such as workflow name. Useful for selecting applicable checks. Defaults to `"default"`. | -| `user_metadata` | `map` | (Optional) Custom metadata to add to any row-level warnings or errors generated by the check. | +| `run_config_name` | `string` | Name of the run config name. Could be any string such as input table or job name (use "default" if not provided). Useful for selecting applicable checks. | +| `user_metadata` | `map` | (Optional) Custom metadata to add to any row-level warnings or errors generated by the check. | Example checks saved in the Delta table: ```python @@ -576,6 +576,8 @@ Example checks saved in the Delta table: +------------------------+-------------+------------------------------------------------------------------------------------------------------------------------+----------+-----------------+-------------------------------------------+ ``` +If `run_config_name` is not provided, "default" is used. Typically, the input table or job name is used for run config name to establish a one-to-one mapping between tables or jobs and checks. + ## Validating syntax of quality checks You can validate the syntax of checks loaded from a storage system or checks defined programmatically before applying them. diff --git a/docs/dqx/docs/guide/quality_checks_storage.mdx b/docs/dqx/docs/guide/quality_checks_storage.mdx index 58fc63988..44018aff5 100644 --- a/docs/dqx/docs/guide/quality_checks_storage.mdx +++ b/docs/dqx/docs/guide/quality_checks_storage.mdx @@ -18,14 +18,14 @@ Saving and loading methods accept a storage backend configuration as input. The * `location`: absolute workspace file path (JSON or YAML). * `TableChecksStorageConfig`: Unity Catalog tables. Containing fields: * `location`: table fully qualified name. - * `run_config_name`: (optional) run configuration name to load (use "default" if not provided). + * `run_config_name`: (optional) run configuration name to load (it can be any string), e.g. input table or job name (use "default" if not provided). * `mode`: (optional) write mode for saving checks (`overwrite` or `append`, default is `overwrite`). The `overwrite` mode will only replace checks for the specific run config and not all checks in the table. * `VolumeFileChecksStorageConfig`: Unity Catalog Volume (JSON/YAML file). Containing fields: * `location`: Unity Catalog Volume file path (JSON or YAML). * `InstallationChecksStorageConfig`: installation-managed location from the run config, ignores location and infers it from `checks_location` in the run config. Containing fields: * `location` (optional): automatically set based on the `checks_location` field from the run configuration. * `install_folder`: (optional) installation folder where DQX is installed, only required when custom installation folder is used. - * `run_config_name` (optional) - run configuration name to load (use "default" if not provided). + * `run_config_name` (optional) - run configuration name to load (it can be any string), e.g. input table or job name (use "default" if not provided). * `product_name`: (optional) name of the product (use "dqx" if not provided). * `assume_user`: (optional) if True, assume user installation, otherwise global installation (skipped if `install_folder` is provided). @@ -72,18 +72,19 @@ If you create checks as a list of `DQRule` objects, you can convert them to meta # save checks as a YAML file in arbitrary workspace location (overwrite the file) using absolute path dq_engine.save_checks(checks, config=WorkspaceFileChecksStorageConfig(location="/Shared/App1/checks.yml")) - # save checks in a Delta table to "default" run config (append checks) + # save checks in a Delta table to "default" run config, append checks dq_engine.save_checks(checks, config=TableChecksStorageConfig(location="catalog.schema.checks_table", mode="append")) - # save checks in a Delta table with specific run config for filtering (overwrite checks) - dq_engine.save_checks(checks, config=TableChecksStorageConfig(location="catalog.schema.checks_table", run_config_name="workflow_001", mode="overwrite")) + # save checks in a Delta table with specific run config for filtering (it can be any string, e.g. input table or job name), overwrite checks + dq_engine.save_checks(checks, config=TableChecksStorageConfig(location="catalog.schema.checks_table", run_config_name="main.default.input_table", mode="overwrite")) # save checks as a YAML in a Unity Catalog Volume location (overwrite the file) dq_engine.save_checks(checks, config=VolumeFileChecksStorageConfig(location="/Volumes/dq/config/checks_volume/App1/checks.yml")) # save checks as a YAML file or table defined in 'checks_location' of the run config # only works if DQX is installed in the workspace - dq_engine.save_checks(checks, config=InstallationChecksStorageConfig(assume_user=True, run_config_name="default")) + # the run config name can be any string, e.g. input table or job name + dq_engine.save_checks(checks, config=InstallationChecksStorageConfig(assume_user=True, run_config_name="main.default.input_table")) ``` @@ -210,15 +211,16 @@ If you create checks as a list of DQRule objects, you can convert them using the # load checks from a Delta table and default run config name checks: list[dict] = dq_engine.load_checks(config=TableChecksStorageConfig(location="catalog.schema.checks_table")) - # load checks from a Delta table with specific run config for filtering - checks: list[dict] = dq_engine.load_checks(config=TableChecksStorageConfig(location="catalog.schema.checks_table", run_config_name="workflow_001")) + # load checks from a Delta table with specific run config for filtering (it can be any string, e.g. input table or job name) + checks: list[dict] = dq_engine.load_checks(config=TableChecksStorageConfig(location="catalog.schema.checks_table", run_config_name="main.default.input_table")) # load checks from a Unity Catalog Volume checks: list[dict] = dq_engine.load_checks(config=VolumeFileChecksStorageConfig(location="/Volumes/dq/config/checks_volume/App1/checks.yml")) # load checks from a file or table defined in the run config ('checks_location' field) # only works if DQX is installed in the workspace - checks: list[dict] = dq_engine.load_checks(config=InstallationChecksStorageConfig(run_config_name="default")) + # the run config name is a string (e.g. input table or job name) + checks: list[dict] = dq_engine.load_checks(config=InstallationChecksStorageConfig(run_config_name="main.default.input_table")) # validate loaded checks assert not dq_engine.validate_checks(checks).has_errors diff --git a/docs/dqx/docs/installation.mdx b/docs/dqx/docs/installation.mdx index aba0971e5..8894f868a 100644 --- a/docs/dqx/docs/installation.mdx +++ b/docs/dqx/docs/installation.mdx @@ -180,7 +180,7 @@ extra_params: # <- optional extra parameters to pass to th run_configs: # <- list of run configurations, each run config defines one specific input and output location -- name: default # <- unique name of the run config (default used during installation) +- name: default # <- unique name of the run config (it can be any string), e.g. input table or job name (use "default" during installation) input_config: # <- optional input data configuration location: s3://iot-ingest/raw # <- input location of the data (table or cloud path) diff --git a/docs/dqx/docs/reference/engine.mdx b/docs/dqx/docs/reference/engine.mdx index 6cbfd5e72..7799db757 100644 --- a/docs/dqx/docs/reference/engine.mdx +++ b/docs/dqx/docs/reference/engine.mdx @@ -49,22 +49,22 @@ The following table outlines the available methods of the `DQEngine` and their f
**Available DQX engine methods** -| Method | Description | Arguments | Supports local execution | -| ----------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------- | -| `apply_checks` | Applies quality checks to the DataFrame and returns a DataFrame with result columns. | `df`: DataFrame to check; `checks`: List of checks defined using DQX classes, each check is an instance of the DQRule class; `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | Yes | -| `apply_checks_and_split` | Applies quality checks to the DataFrame and returns valid and invalid (quarantine) DataFrames with result columns. | `df`: DataFrame to check; `checks`: List of checks defined using DQX classes, each check is an instance of the DQRule class; `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | Yes | -| `apply_checks_and_save_in_table` | Applies quality checks using DQRule objects and writes results to valid and invalid Delta table(s) with result columns. | `input_config`: `InputConfig` object with the table name and options for reading the input data; `checks`: List of checks defined using DQX classes, each check is an instance of the DQRule class; `output_config`: `OutputConfig` object with the table name, output mode, and options for the output data; `quarantine_config`: `OutputConfig` object with the table name, output mode, and options for the quarantine data - if provided, data will be split; `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | No | -| `apply_checks_by_metadata` | Applies quality checks defined as a dictionary to the DataFrame and returns a DataFrame with result columns. | `df`: DataFrame to check; `checks`: List of checks defined as dictionary; `custom_check_functions`: (optional) Dictionary with custom check functions (e.g., globals() of the calling module); `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | Yes | -| `apply_checks_by_metadata_and_split` | Applies quality checks defined as a dictionary and returns valid and invalid (quarantine) DataFrames. | `df`: DataFrame to check; `checks`: List of checks defined as dictionary; `custom_check_functions`: (optional) Dictionary with custom check functions (e.g., globals() of the calling module); `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | Yes | -| `apply_checks_by_metadata_and_save_in_table` | Applies quality checks defined as a dictionary and writes results to valid and invalid Delta table(s) with result columns. | `input_config`: `InputConfig` object with the table name and options for reading the input data; `checks`: List of checks defined as dictionary; `output_config`: `OutputConfig` object with the table name, output mode, and options for the output data; `quarantine_config`: `OutputConfig` object with the table name, output mode, and options for the quarantine data - if provided, data will be split; `custom_check_functions`: (optional) Dictionary with custom check functions; `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | No | -| `apply_checks_and_save_in_tables` | Applies quality checks persisted in a storage to multiple tables and writes results to valid and invalid Delta table(s) with result columns. | `run_configs`: list of run config objects (`RunConfig`) containing input config (`InputConfig`), output config (`OutputConfig`), quarantine config (`OutputConfig`, if provided data will be split), 'checks_location', and if provided 'reference_tables' and 'custom_check_functions'; `max_parallelism`: (optional) Maximum number of tables to check in parallel (defaults to the number of CPU cores). | No | +| Method | Description | Arguments | Supports local execution | +| ----------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------- | +| `apply_checks` | Applies quality checks to the DataFrame and returns a DataFrame with result columns. | `df`: DataFrame to check; `checks`: List of checks defined using DQX classes, each check is an instance of the DQRule class; `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | Yes | +| `apply_checks_and_split` | Applies quality checks to the DataFrame and returns valid and invalid (quarantine) DataFrames with result columns. | `df`: DataFrame to check; `checks`: List of checks defined using DQX classes, each check is an instance of the DQRule class; `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | Yes | +| `apply_checks_and_save_in_table` | Applies quality checks using DQRule objects and writes results to valid and invalid Delta table(s) with result columns. | `input_config`: `InputConfig` object with the table name and options for reading the input data; `checks`: List of checks defined using DQX classes, each check is an instance of the DQRule class; `output_config`: `OutputConfig` object with the table name, output mode, and options for the output data; `quarantine_config`: `OutputConfig` object with the table name, output mode, and options for the quarantine data - if provided, data will be split; `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | No | +| `apply_checks_by_metadata` | Applies quality checks defined as a dictionary to the DataFrame and returns a DataFrame with result columns. | `df`: DataFrame to check; `checks`: List of checks defined as dictionary; `custom_check_functions`: (optional) Dictionary with custom check functions (e.g., globals() of the calling module); `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | Yes | +| `apply_checks_by_metadata_and_split` | Applies quality checks defined as a dictionary and returns valid and invalid (quarantine) DataFrames. | `df`: DataFrame to check; `checks`: List of checks defined as dictionary; `custom_check_functions`: (optional) Dictionary with custom check functions (e.g., globals() of the calling module); `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | Yes | +| `apply_checks_by_metadata_and_save_in_table` | Applies quality checks defined as a dictionary and writes results to valid and invalid Delta table(s) with result columns. | `input_config`: `InputConfig` object with the table name and options for reading the input data; `checks`: List of checks defined as dictionary; `output_config`: `OutputConfig` object with the table name, output mode, and options for the output data; `quarantine_config`: `OutputConfig` object with the table name, output mode, and options for the quarantine data - if provided, data will be split; `custom_check_functions`: (optional) Dictionary with custom check functions; `ref_dfs`: (optional) Reference dataframes to use in the checks, if applicable. | No | +| `apply_checks_and_save_in_tables` | Applies quality checks persisted in a storage to multiple tables and writes results to valid and invalid Delta table(s) with result columns. | `run_configs`: list of run config objects (`RunConfig`) containing input config (`InputConfig`), output config (`OutputConfig`), quarantine config (`OutputConfig`, if provided data will be split), 'checks_location', and if provided 'reference_tables' and 'custom_check_functions'; `max_parallelism`: (optional) Maximum number of tables to check in parallel (defaults to the number of CPU cores). | No | | `apply_checks_and_save_in_tables_for_patterns` | Applies quality checks persisted in a storage to multiple tables matching provided wildcard patterns and writes results to valid and invalid Delta table(s) with result columns. Skip output and quarantine tables based on specified suffixes. | `patterns`: List of table names or filesystem-style wildcards (e.g. 'schema.*') to include (if None, all tables are included); ; `exclude_patterns`: (optional) List of table names or filesystem-style wildcards (e.g. '*_dq_output') to exclude, useful if wanting to exclude existing output or quarantine tables; `checks_location`: Location of the checks files (e.g. absolute workspace or volume directory or delta table), for file based locations, checks are expected to be found under 'checks_location/<input_table>.yml'; `exclude_matched`:(optional) Whether to exclude matched tables (default False); `run_config_template`: (optional) Run configuration template to use for all tables (skip location in the 'input_config', 'output_config', and 'quarantine_config' fields as it is derived from patterns, skip 'checks_location' of the run config as it is derived separately, autogenerate 'input_config' and 'output_config' if not provided, use 'reference_tables' and 'custom_check_functions' if provided; `max_parallelism`: (optional) Maximum number of tables to check in parallel (defaults to the number of CPU cores); `output_table_suffix`: (optional) Suffix to append to the output table name (default "_dq_output"); `quarantine_table_suffix`: (optional) Suffix to append to the quarantine table name (default "_dq_quarantine"). | No | -| `validate_checks` | Validates the provided quality checks to ensure they conform to the expected structure and types. | `checks`: List of checks to validate; `custom_check_functions`: (optional) Dictionary of custom check functions that can be used; `validate_custom_check_functions`: (optional) If set to True, validates custom check functions (defaults to True). | Yes | -| `get_invalid` | Retrieves records from the DataFrame that violate data quality checks (records with warnings and errors). | `df`: Input DataFrame. | Yes | -| `get_valid` | Retrieves records from the DataFrame that pass all data quality checks. | `df`: Input DataFrame. | Yes | -| `load_checks` | Loads quality rules (checks) from storage backend. Multiple storage backends are supported including tables, files or workspace files, installation-managed sources where the location is inferred automatically from run config. | `config`: Configuration for loading checks from a storage backend, i.e. `FileChecksStorageConfig`: file in a local filesystem (YAML or JSON), or workspace files if invoked from Databricks notebook or job; `WorkspaceFileChecksStorageConfig`: file in a workspace (YAML or JSON) using absolute paths; `VolumeFileChecksStorageConfig`: file in a Unity Catalog Volume (YAML or JSON); `TableChecksStorageConfig`: a table; `InstallationChecksStorageConfig`: installation-managed storage backend, using the `checks_location` field from the run configuration. See more details below. | Yes (only with `FileChecksStorageConfig`) | -| `save_checks` | Saves quality rules (checks) to storage backend. Multiple storage backends are supported including tables, files or workspace files, installation-managed targets where the location is inferred automatically from run config. | `checks`: List of checks defined as dictionary; `config`: Configuration for saving checks in a storage backend, i.e. `FileChecksStorageConfig`: file in a local filesystem (YAML or JSON), or workspace files if invoked from Databricks notebook or job; `WorkspaceFileChecksStorageConfig`: file in a workspace (YAML or JSON); `VolumeFileChecksStorageConfig`: file in a Unity Catalog Volume (YAML or JSON); `TableChecksStorageConfig`: a table; `InstallationChecksStorageConfig`: storage defined in the installation context, using the `checks_location` field from the run configuration. See more details below. | Yes (only with `FileChecksStorageConfig`) | -| `save_results_in_table` | Save quality checking results in delta table(s). | `output_df`: (optional) Dataframe containing the output data; `quarantine_df`: (optional) Dataframe containing the output data; `output_config`: `OutputConfig` object with the table name, output mode, and options for the output data; `quarantine_config`: `OutputConfig` object with the table name, output mode, and options for the quarantine data - if provided, data will be split; `run_config_name`: Name of the run config to use; `install_folder`: (optional) installation folder where DQX is installed, only required when custom installation folder is used; `assume_user`: (optional) If True, assume user installation, otherwise global installation (skipped if `install_folder` is provided). | No | +| `validate_checks` | Validates the provided quality checks to ensure they conform to the expected structure and types. | `checks`: List of checks to validate; `custom_check_functions`: (optional) Dictionary of custom check functions that can be used; `validate_custom_check_functions`: (optional) If set to True, validates custom check functions (defaults to True). | Yes | +| `get_invalid` | Retrieves records from the DataFrame that violate data quality checks (records with warnings and errors). | `df`: Input DataFrame. | Yes | +| `get_valid` | Retrieves records from the DataFrame that pass all data quality checks. | `df`: Input DataFrame. | Yes | +| `load_checks` | Loads quality rules (checks) from storage backend. Multiple storage backends are supported including tables, files or workspace files, installation-managed sources where the location is inferred automatically from run config. | `config`: Configuration for loading checks from a storage backend, i.e. `FileChecksStorageConfig`: file in a local filesystem (YAML or JSON), or workspace files if invoked from Databricks notebook or job; `WorkspaceFileChecksStorageConfig`: file in a workspace (YAML or JSON) using absolute paths; `VolumeFileChecksStorageConfig`: file in a Unity Catalog Volume (YAML or JSON); `TableChecksStorageConfig`: a table; `InstallationChecksStorageConfig`: installation-managed storage backend, using the `checks_location` field from the run configuration. See more details below. | Yes (only with `FileChecksStorageConfig`) | +| `save_checks` | Saves quality rules (checks) to storage backend. Multiple storage backends are supported including tables, files or workspace files, installation-managed targets where the location is inferred automatically from run config. | `checks`: List of checks defined as dictionary; `config`: Configuration for saving checks in a storage backend, i.e. `FileChecksStorageConfig`: file in a local filesystem (YAML or JSON), or workspace files if invoked from Databricks notebook or job; `WorkspaceFileChecksStorageConfig`: file in a workspace (YAML or JSON); `VolumeFileChecksStorageConfig`: file in a Unity Catalog Volume (YAML or JSON); `TableChecksStorageConfig`: a table; `InstallationChecksStorageConfig`: storage defined in the installation context, using the `checks_location` field from the run configuration. See more details below. | Yes (only with `FileChecksStorageConfig`) | +| `save_results_in_table` | Save quality checking results in delta table(s). | `output_df`: (optional) Dataframe containing the output data; `quarantine_df`: (optional) Dataframe containing the output data; `output_config`: `OutputConfig` object with the table name, output mode, and options for the output data; `quarantine_config`: `OutputConfig` object with the table name, output mode, and options for the quarantine data - if provided, data will be split; `run_config_name`: Name of the run config to use (it can be any string), e.g. input table or job name (use "default" if not provided); `install_folder`: (optional) installation folder where DQX is installed, only required when custom installation folder is used; `assume_user`: (optional) If True, assume user installation, otherwise global installation (skipped if `install_folder` is provided). | No | The 'Supports local execution' in the above table indicates which methods can be used for local testing without a Databricks workspace (see the usage in [local testing section](/docs/reference/testing/#local-execution-and-testing-with-dqengine)). @@ -101,14 +101,14 @@ Supported storage backend configurations (implementations of `BaseChecksStorageC * `location`: absolute workspace file path (JSON or YAML). * `TableChecksStorageConfig` can be used to save or load checks from a table, with fields: * `location`: table fully qualified name. - * `run_config_name`: (optional) run configuration name to load (use "default" if not provided). + * `run_config_name`: (optional) run configuration name to load (it can be any string), e.g. input table or job name (use "default" if not provided). * `mode`: (optional) write mode for saving checks (`overwrite` or `append`, default is `overwrite`). The `overwrite` mode will only replace checks for the specific run config and not all checks in the table. * `VolumeFileChecksStorageConfig` can be used to save or load checks from a Unity Catalog Volume file, with fields: * `location`: Unity Catalog Volume file path (JSON or YAML). * `InstallationChecksStorageConfig` can be used to save or load checks from workspace installation, with fields: * `location` (optional): automatically set based on the `checks_location` field from the run configuration. * `install_folder`: (optional) installation folder where DQX is installed, only required when custom installation folder is used. - * `run_config_name` (optional) - run configuration name to load (use "default" if not provided). + * `run_config_name` (optional) - run configuration name to load (it can be any string), e.g. input table or job name (use "default" if not provided). * `product_name`: (optional) name of the product (use "dqx" if not provided). * `assume_user`: (optional) if True, assume user installation, otherwise global installation (skipped if `install_folder` is provided). diff --git a/docs/dqx/docs/reference/quality_checks.mdx b/docs/dqx/docs/reference/quality_checks.mdx index 5020e6b3d..3b8a6c750 100644 --- a/docs/dqx/docs/reference/quality_checks.mdx +++ b/docs/dqx/docs/reference/quality_checks.mdx @@ -11,6 +11,8 @@ import TabItem from '@theme/TabItem'; This page provides a reference for the quality check functions available in DQX. The terms `checks` and `rules` are used interchangeably in the documentation, as they are synonymous in DQX. +All DQX rules are evaluated independently and in parallel. If one rule fails, the others are still processed. + You can explore the implementation details of the check functions [here](https://github.com/databrickslabs/dqx/blob/v0.9.3/src/databricks/labs/dqx/check_funcs.py). ## Row-level checks reference @@ -20,50 +22,50 @@ You can also define your own custom checks (see [Creating custom checks](#creati
**Available row-level checks** -| Check | Description | Arguments | -| ---------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `is_not_null` | Checks whether the values in the input column are not null. | `column`: column to check (can be a string column name or a column expression) | -| `is_not_empty` | Checks whether the values in the input column are not empty (but may be null). | `column`: column to check (can be a string column name or a column expression) | -| `is_not_null_and_not_empty` | Checks whether the values in the input column are not null and not empty. | `column`: column to check (can be a string column name or a column expression); `trim_strings`: optional boolean flag to trim spaces from strings | -| `is_in_list` | Checks whether the values in the input column are present in the list of allowed values (null values are allowed). This check is not suited for large lists of allowed values. In such cases, it’s recommended to use the `foreign_key` dataset-level check instead. | `column`: column to check (can be a string column name or a column expression); `allowed`: list of allowed values | -| `is_not_null_and_is_in_list` | Checks whether the values in the input column are not null and present in the list of allowed values. This check is not suited for large lists of allowed values. In such cases, it’s recommended to use the `foreign_key` dataset-level check instead. | `column`: column to check (can be a string column name or a column expression); `allowed`: list of allowed values | -| `is_not_null_and_not_empty_array` | Checks whether the values in the array input column are not null and not empty. | `column`: column to check (can be a string column name or a column expression) | -| `is_in_range` | Checks whether the values in the input column are in the provided range (inclusive of both boundaries). | `column`: column to check (can be a string column name or a column expression); `min_limit`: min limit as number, date, timestamp, column name or sql expression; `max_limit`: max limit as number, date, timestamp, column name or sql expression | -| `is_not_in_range` | Checks whether the values in the input column are outside the provided range (inclusive of both boundaries). | `column`: column to check (can be a string column name or a column expression); `min_limit`: min limit as number, date, timestamp, column name or sql expression; `max_limit`: max limit as number, date, timestamp, column name or sql expression | -| `is_not_equal_to` | Checks whether the values in the input column are not equal to the provided value. | `column`: column to check (can be a string column name or a column expression); `value`: comparison value as number, string, date, timestamp, column name, or SQL expression | -| `is_equal_to` | Checks whether the values in the input column are equal to the provided value. | `column`: column to check (can be a string column name or a column expression); `value`: comparison value as number, string, date, timestamp, column name, or SQL expression | -| `is_not_less_than` | Checks whether the values in the input column are not less than the provided limit. | `column`: column to check (can be a string column name or a column expression); `limit`: limit as number, date, timestamp, column name or sql expression | -| `is_not_greater_than` | Checks whether the values in the input column are not greater than the provided limit. | `column`: column to check (can be a string column name or a column expression); `limit`: limit as number, date, timestamp, column name or sql expression | -| `is_valid_date` | Checks whether the values in the input column have valid date formats. | `column`: column to check (can be a string column name or a column expression); `date_format`: optional date format (e.g. 'yyyy-mm-dd') | -| `is_valid_timestamp` | Checks whether the values in the input column have valid timestamp formats. | `column`: column to check (can be a string column name or a column expression); `timestamp_format`: optional timestamp format (e.g. 'yyyy-mm-dd HH:mm:ss') | -| `is_not_in_future` | Checks whether the values in the input column contain a timestamp that is not in the future, where 'future' is defined as current_timestamp + offset (in seconds). | `column`: column to check (can be a string column name or a column expression); `offset`: offset to use; `curr_timestamp`: current timestamp, if not provided current_timestamp() function is used | -| `is_not_in_near_future` | Checks whether the values in the input column contain a timestamp that is not in the near future, where 'near future' is defined as greater than the current timestamp but less than the current_timestamp + offset (in seconds). | `column`: column to check (can be a string column name or a column expression); `offset`: offset to use; `curr_timestamp`: current timestamp, if not provided current_timestamp() function is used | -| `is_older_than_n_days` | Checks whether the values in one input column are at least N days older than the values in another column. | `column`: column to check (can be a string column name or a column expression); `days`: number of days; `curr_date`: current date, if not provided current_date() function is used; `negate`: if the condition should be negated | -| `is_older_than_col2_for_n_days` | Checks whether the values in one input column are at least N days older than the values in another column. | `column1`: first column to check (can be a string column name or a column expression); `column2`: second column to check (can be a string column name or a column expression); `days`: number of days; `negate`: if the condition should be negated | -| `regex_match` | Checks whether the values in the input column match a given regex. | `column`: column to check (can be a string column name or a column expression); regex: regex to check; `negate`: if the condition should be negated (true) or not | -| `is_valid_ipv4_address` | Checks whether the values in the input column have valid IPv4 address format. | `column` to check (can be a string column name or a column expression) | -| `is_ipv4_address_in_cidr` | Checks whether the values in the input column have valid IPv4 address format and fall within the given CIDR block. | `column`: column to check (can be a string column name or a column expression); `cidr_block`: CIDR block string | -| `is_valid_ipv6_address` | Checks whether the values in the input column have valid IPv6 address format. | `column` to check (can be a string column name or a column expression) | -| `is_ipv6_address_in_cidr` | Checks whether the values in the input column have valid IPv6 address format and fall within the given CIDR block. | `column`: column to check (can be a string column name or a column expression); `cidr_block`: CIDR block string | -| `sql_expression` | Checks whether the values meet the condition provided as an SQL expression, e.g. `a = 'str1' and a > b`. SQL expressions are evaluated at runtime, so ensure that the expression is safe and that functions used within it (e.g. h3_ischildof, division) do not throw exceptions. You can achieve this by validating input arguments or columns beforehand using guards such as CASE WHEN, IS NOT NULL, RLIKE, or type try casts. | `expression`: sql expression to check on a DataFrame (fail the check if expression evaluates to False, pass if it evaluates to True); `msg`: optional message to output; `name`: optional name of the resulting column (it can be overwritten by `name` specified at the check level); `negate`: if the condition should be negated; `columns`: optional list of columns to be used for reporting and as name prefix if name not provided, unused in the actual logic | -| `is_data_fresh` | Checks whether the values in the input timestamp column are not older than the specified number of minutes from the base timestamp column. This is useful for identifying stale data due to delayed pipelines and helps catch upstream issues early. | `column`: column of type timestamp/date to check (can be a string column name or a column expression); `max_age_minutes`: maximum age in minutes before data is considered stale; `base_timestamp`: optional base timestamp column from which the stale check is calculated. This can be a string, column expression, datetime value or literal value ex:F.lit(datetime(2024,1,1)). If not provided current_timestamp() function is used | -| `does_not_contain_pii` | Checks whether the values in the input column contain Personally Identifiable Information (PII). Uses Microsoft Presidio to detect various named entities (e.g. PERSON, ADDRESS, EMAIL_ADDRESS). Requires installation of PII detection extras: `pip install 'databricks-labs-dqx[pii-detection]'`. See more details [here](#detecting-personally-identifiable-information-pii). | `column`: column to check (can be a string column name or a column expression); `threshold`: confidence threshold for PII detection (0.0 to 1.0, default: 0.7); `language`: optional language of the text (default: 'en'); `entities`: optional list of entities to detect; `nlp_engine_config`: optional dictionary configuring the NLP engine used for PII detection, see the [Presidio documentation](https://microsoft.github.io/presidio/analyzer/customizing_nlp_models/) for more information | -| `is_latitude` | Checks whether the values in the input column are valid latitude values (i.e. between -90 and 90 degrees). | `column`: column to check (can be a string column name or a column expression) | -| `is_longitude` | Checks whether the values in the input column are valid longitude values (i.e. between -180 and 180 degrees). | `column`: column to check (can be a string column name or a column expression) | -| `is_geometry` | Checks whether the values in the input column are valid geometries. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | -| `is_geography` | Checks whether the values in the input column are valid geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | -| `is_point` | Checks whether the values in the input column are point geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | -| `is_linestring` | Checks whether the values in the input column are linestring geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | -| `is_polygon` | Checks whether the values in the input column are polygon geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | -| `is_multipoint` | Checks whether the values in the input column are multipoint geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | -| `is_multilinestring` | Checks whether the values in the input column are multilinestring geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | -| `is_multipolygon` | Checks whether the values in the input column are multipolygon geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | -| `is_geometrycollection` | Checks whether the values in the input column are geometrycollection geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | -| `is_ogc_valid` | Checks whether the values in the input column are valid geometries in the OGC sense. I.e a bowtie polygon is invalid because it has a self intersection. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | -| `is_non_empty_geometry` | Checks whether the values in the input column are non-empty geometries. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | -| `has_dimension` | Checks whether the values in the input column are geometries of the specified dimension (2D projected dimension). This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression); `dimension`: dimension to check | -| `has_x_coordinate_between` | Checks whether the values in the input column are geometries with x coordinate between the provided boundaries. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression); `min_value`: minimum value; `max_value`: maximum value | -| `has_y_coordinate_between` | Checks whether the values in the input column are geometries with y coordinate between the provided boundaries. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression); `min_value`: minimum value; `max_value`: maximum value | | `column`: column to check (can be a string column name or a column expression); `min_value`: minimum value; `max_value`: maximum value | +| Check | Description | Arguments | +| ---------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `is_not_null` | Checks whether the values in the input column are not null. | `column`: column to check (can be a string column name or a column expression) | +| `is_not_empty` | Checks whether the values in the input column are not empty (but may be null). | `column`: column to check (can be a string column name or a column expression) | +| `is_not_null_and_not_empty` | Checks whether the values in the input column are not null and not empty. | `column`: column to check (can be a string column name or a column expression); `trim_strings`: optional boolean flag to trim spaces from strings | +| `is_in_list` | Checks whether the values in the input column are present in the list of allowed values (null values are allowed). This check is not suited for large lists of allowed values. In such cases, it’s recommended to use the `foreign_key` dataset-level check instead. | `column`: column to check (can be a string column name or a column expression); `allowed`: list of allowed values | +| `is_not_null_and_is_in_list` | Checks whether the values in the input column are not null and present in the list of allowed values. This check is not suited for large lists of allowed values. In such cases, it’s recommended to use the `foreign_key` dataset-level check instead. | `column`: column to check (can be a string column name or a column expression); `allowed`: list of allowed values | +| `is_not_null_and_not_empty_array` | Checks whether the values in the array input column are not null and not empty. | `column`: column to check (can be a string column name or a column expression) | +| `is_in_range` | Checks whether the values in the input column are in the provided range (inclusive of both boundaries). | `column`: column to check (can be a string column name or a column expression); `min_limit`: min limit as number, date, timestamp, column name or sql expression; `max_limit`: max limit as number, date, timestamp, column name or sql expression | +| `is_not_in_range` | Checks whether the values in the input column are outside the provided range (inclusive of both boundaries). | `column`: column to check (can be a string column name or a column expression); `min_limit`: min limit as number, date, timestamp, column name or sql expression; `max_limit`: max limit as number, date, timestamp, column name or sql expression | +| `is_not_equal_to` | Checks whether the values in the input column are not equal to the provided value. | `column`: column to check (can be a string column name or a column expression); `value`: comparison value as number, string, date, timestamp, column name, or SQL expression | +| `is_equal_to` | Checks whether the values in the input column are equal to the provided value. | `column`: column to check (can be a string column name or a column expression); `value`: comparison value as number, string, date, timestamp, column name, or SQL expression | +| `is_not_less_than` | Checks whether the values in the input column are not less than the provided limit. | `column`: column to check (can be a string column name or a column expression); `limit`: limit as number, date, timestamp, column name or sql expression | +| `is_not_greater_than` | Checks whether the values in the input column are not greater than the provided limit. | `column`: column to check (can be a string column name or a column expression); `limit`: limit as number, date, timestamp, column name or sql expression | +| `is_valid_date` | Checks whether the values in the input column have valid date formats. | `column`: column to check (can be a string column name or a column expression); `date_format`: optional date format (e.g. 'yyyy-mm-dd') | +| `is_valid_timestamp` | Checks whether the values in the input column have valid timestamp formats. | `column`: column to check (can be a string column name or a column expression); `timestamp_format`: optional timestamp format (e.g. 'yyyy-mm-dd HH:mm:ss') | +| `is_not_in_future` | Checks whether the values in the input column contain a timestamp that is not in the future, where 'future' is defined as current_timestamp + offset (in seconds). | `column`: column to check (can be a string column name or a column expression); `offset`: offset to use; `curr_timestamp`: current timestamp, if not provided current_timestamp() function is used | +| `is_not_in_near_future` | Checks whether the values in the input column contain a timestamp that is not in the near future, where 'near future' is defined as greater than the current timestamp but less than the current_timestamp + offset (in seconds). | `column`: column to check (can be a string column name or a column expression); `offset`: offset to use; `curr_timestamp`: current timestamp, if not provided current_timestamp() function is used | +| `is_older_than_n_days` | Checks whether the values in one input column are at least N days older than the values in another column. | `column`: column to check (can be a string column name or a column expression); `days`: number of days; `curr_date`: current date, if not provided current_date() function is used; `negate`: if the condition should be negated | +| `is_older_than_col2_for_n_days` | Checks whether the values in one input column are at least N days older than the values in another column. | `column1`: first column to check (can be a string column name or a column expression); `column2`: second column to check (can be a string column name or a column expression); `days`: number of days; `negate`: if the condition should be negated | +| `regex_match` | Checks whether the values in the input column match a given regex. | `column`: column to check (can be a string column name or a column expression); regex: regex to check; `negate`: if the condition should be negated (true) or not | +| `is_valid_ipv4_address` | Checks whether the values in the input column have valid IPv4 address format. | `column` to check (can be a string column name or a column expression) | +| `is_ipv4_address_in_cidr` | Checks whether the values in the input column have valid IPv4 address format and fall within the given CIDR block. | `column`: column to check (can be a string column name or a column expression); `cidr_block`: CIDR block string | +| `is_valid_ipv6_address` | Checks whether the values in the input column have valid IPv6 address format. | `column` to check (can be a string column name or a column expression) | +| `is_ipv6_address_in_cidr` | Checks whether the values in the input column have valid IPv6 address format and fall within the given CIDR block. | `column`: column to check (can be a string column name or a column expression); `cidr_block`: CIDR block string | +| `sql_expression` | Checks whether the values meet the condition provided as an SQL expression, e.g. `a = 'str1' and a > b`. SQL expressions are evaluated at runtime, so ensure that the expression is safe and that functions used within it (e.g. h3_ischildof, division) do not throw exceptions. You can achieve this by validating input arguments or columns beforehand using guards such as CASE WHEN, IS NOT NULL, RLIKE, or type try casts. | `expression`: sql expression to check on a DataFrame (fail the check if expression evaluates to False, pass if it evaluates to True); `msg`: optional message to output; `name`: optional name of the resulting column (it can be overwritten by `name` specified at the check level); `negate`: if the condition should be negated; `columns`: optional list of columns used in the sql expression to validate they can be resolved in the input DataFrame, also used for reporting purposes and as a name prefix when a check name is not provided | +| `is_data_fresh` | Checks whether the values in the input timestamp column are not older than the specified number of minutes from the base timestamp column. This is useful for identifying stale data due to delayed pipelines and helps catch upstream issues early. | `column`: column of type timestamp/date to check (can be a string column name or a column expression); `max_age_minutes`: maximum age in minutes before data is considered stale; `base_timestamp`: optional base timestamp column from which the stale check is calculated. This can be a string, column expression, datetime value or literal value ex:F.lit(datetime(2024,1,1)). If not provided current_timestamp() function is used | +| `does_not_contain_pii` | Checks whether the values in the input column contain Personally Identifiable Information (PII). Uses Microsoft Presidio to detect various named entities (e.g. PERSON, ADDRESS, EMAIL_ADDRESS). Requires installation of PII detection extras: `pip install 'databricks-labs-dqx[pii-detection]'`. See more details [here](#detecting-personally-identifiable-information-pii). | `column`: column to check (can be a string column name or a column expression); `threshold`: confidence threshold for PII detection (0.0 to 1.0, default: 0.7); `language`: optional language of the text (default: 'en'); `entities`: optional list of entities to detect; `nlp_engine_config`: optional dictionary configuring the NLP engine used for PII detection, see the [Presidio documentation](https://microsoft.github.io/presidio/analyzer/customizing_nlp_models/) for more information | +| `is_latitude` | Checks whether the values in the input column are valid latitude values (i.e. between -90 and 90 degrees). | `column`: column to check (can be a string column name or a column expression) | +| `is_longitude` | Checks whether the values in the input column are valid longitude values (i.e. between -180 and 180 degrees). | `column`: column to check (can be a string column name or a column expression) | +| `is_geometry` | Checks whether the values in the input column are valid geometries. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | +| `is_geography` | Checks whether the values in the input column are valid geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | +| `is_point` | Checks whether the values in the input column are point geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | +| `is_linestring` | Checks whether the values in the input column are linestring geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | +| `is_polygon` | Checks whether the values in the input column are polygon geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | +| `is_multipoint` | Checks whether the values in the input column are multipoint geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | +| `is_multilinestring` | Checks whether the values in the input column are multilinestring geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | +| `is_multipolygon` | Checks whether the values in the input column are multipolygon geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | +| `is_geometrycollection` | Checks whether the values in the input column are geometrycollection geometries/geographies. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | +| `is_ogc_valid` | Checks whether the values in the input column are valid geometries in the OGC sense. I.e a bowtie polygon is invalid because it has a self intersection. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | +| `is_non_empty_geometry` | Checks whether the values in the input column are non-empty geometries. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression) | +| `has_dimension` | Checks whether the values in the input column are geometries of the specified dimension (2D projected dimension). This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression); `dimension`: dimension to check | +| `has_x_coordinate_between` | Checks whether the values in the input column are geometries with x coordinate between the provided boundaries. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression); `min_value`: minimum value; `max_value`: maximum value | +| `has_y_coordinate_between` | Checks whether the values in the input column are geometries with y coordinate between the provided boundaries. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression); `min_value`: minimum value; `max_value`: maximum value | | `column`: column to check (can be a string column name or a column expression); `min_value`: minimum value; `max_value`: maximum value |
@@ -2218,8 +2220,9 @@ SQL Expressions are also useful if you need to make cross-column validation, for ### Using Python function -If you need a reusable check or want to implement more complex logic which is challenging to implement using SQL expression, you can define your own custom check function in Python. +If you need a reusable check or want to implement more complex logic which may be challenging to implement using SQL expression, you can define your own custom check function in Python. A check function is a callable that uses `make_condition` to return `pyspark.sql.Column`. +You should make sure to handle edge cases as part of your custom check (e.g. properly handle Null values). #### Custom check example diff --git a/pyproject.toml b/pyproject.toml index 074eaff98..95649f2bb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -529,7 +529,7 @@ indent-string = " " max-line-length = 100 # Maximum number of lines in a module. -max-module-lines = 8000 +max-module-lines = 10000 # Allow the body of a class to be on the same line as the declaration if body # contains single statement. diff --git a/src/databricks/labs/dqx/check_funcs.py b/src/databricks/labs/dqx/check_funcs.py index 6b701274f..962d550a8 100644 --- a/src/databricks/labs/dqx/check_funcs.py +++ b/src/databricks/labs/dqx/check_funcs.py @@ -227,7 +227,8 @@ def sql_expression( name: optional name of the resulting column, automatically generated if None negate: if the condition should be negated (true) or not. For example, "col is not null" will mark null values as "bad". Although sometimes it's easier to specify it other way around "col is null" + negate set to True - columns: optional list of columns to be used for reporting. Unused in the actual logic. + columns: optional list of columns to be used for validation against the actual input DataFrame, + reporting and for constructing name prefix if check name is not provided. Returns: new Column diff --git a/src/databricks/labs/dqx/checks_serializer.py b/src/databricks/labs/dqx/checks_serializer.py index 3d9c8150c..d98c32dd3 100644 --- a/src/databricks/labs/dqx/checks_serializer.py +++ b/src/databricks/labs/dqx/checks_serializer.py @@ -49,13 +49,13 @@ def serialize_checks_from_dataframe(df: DataFrame, run_config_name: str = "defau Args: df: DataFrame with data quality check rules. Each row should define a check. Rows should have the following columns: - - *name* - Name that will be given to a resulting column. Autogenerated if not provided - - *criticality* (optional) - Possible values are *error* (data going only into "bad" dataframe) and *warn* (data is going into both dataframes) - - *check* - DQX check function used in the check; A *StructType* column defining the data quality check - - *filter* - Expression for filtering data quality checks - - *run_config_name* (optional) - Run configuration name for storing checks across runs + - *name* - Name that will be given to a resulting column. Autogenerated if not provided. + - *criticality* (optional) - Possible values are *error* (data going only into "bad" dataframe) and *warn* (data is going into both dataframes). + - *check* - DQX check function used in the check; A *StructType* column defining the data quality check. + - *filter* - Expression for filtering data quality checks. + - *run_config_name* (optional) - Run configuration name for storing checks across runs. - *user_metadata* (optional) - User-defined key-value pairs added to metadata generated by the check. - run_config_name: Run configuration name for filtering quality rules + run_config_name: Run configuration name for filtering quality rules, e.g. input table or job name (use "default" if not provided). Returns: List of data quality check specifications as a Python dictionary @@ -111,7 +111,7 @@ def deserialize_checks_to_dataframe( (data is going into both dataframes) - *filter* (optional) - Expression for filtering data quality checks - *user_metadata* (optional) - User-defined key-value pairs added to metadata generated by the check. - run_config_name: Run configuration name for storing quality checks across runs + run_config_name: Run configuration name for storing quality checks across runs, e.g. input table or job name (use "default" if not provided) Returns: DataFrame with data quality check rules diff --git a/src/databricks/labs/dqx/checks_storage.py b/src/databricks/labs/dqx/checks_storage.py index f8e308df7..7541d97ab 100644 --- a/src/databricks/labs/dqx/checks_storage.py +++ b/src/databricks/labs/dqx/checks_storage.py @@ -388,8 +388,8 @@ def create_for_location( Abstract method to create a handler and config based on checks location. Args: - location: location of the checks (file path, table name, volume, etc.) - run_config_name: the name of the run configuration to use for checks (default is 'default'). + location: location of the checks (file path, table name, volume, etc.). + run_config_name: the name of the run configuration to use for checks, e.g. input table or job name (use "default" if not provided). Returns: An instance of the corresponding BaseChecksStorageHandler. diff --git a/src/databricks/labs/dqx/config.py b/src/databricks/labs/dqx/config.py index d1fe902ca..fd62cafba 100644 --- a/src/databricks/labs/dqx/config.py +++ b/src/databricks/labs/dqx/config.py @@ -115,7 +115,7 @@ def get_run_config(self, run_config_name: str | None = "default") -> RunConfig: """Get the run configuration for a given run name, or the default configuration if no run name is provided. Args: - run_config_name: The name of the run configuration to get. + run_config_name: The name of the run configuration to get, e.g. input table or job name (use "default" if not provided). Returns: The run configuration. @@ -181,7 +181,7 @@ class TableChecksStorageConfig(BaseChecksStorageConfig): Args: location: The table name where the checks are stored. - run_config_name: The name of the run configuration to use for checks (default is 'default'). + run_config_name: The name of the run configuration to use for checks, e.g. input table or job name (use "default" if not provided). mode: The mode for writing checks to a table (e.g., 'append' or 'overwrite'). The *overwrite* mode will only replace checks for the specific run config and not all checks in the table. """ @@ -222,7 +222,7 @@ class InstallationChecksStorageConfig( location: The installation path where the checks are stored (e.g., table name, file path). Not used when using installation method, as it is retrieved from the installation config, unless overwrite_location is enabled. - run_config_name: The name of the run configuration to use for checks (default is 'default'). + run_config_name: The name of the run configuration to use for checks, e.g. input table or job name (use "default" if not provided). product_name: The product name for retrieving checks from the installation (default is 'dqx'). assume_user: Whether to assume the user is the owner of the checks (default is True). install_folder: The installation folder where DQX is installed. diff --git a/src/databricks/labs/dqx/config_loader.py b/src/databricks/labs/dqx/config_loader.py index 88549c534..e300c80ad 100644 --- a/src/databricks/labs/dqx/config_loader.py +++ b/src/databricks/labs/dqx/config_loader.py @@ -35,10 +35,10 @@ def load_run_config( Load run configuration from the installation. Args: - run_config_name: Name of the run configuration to use. + run_config_name: Name of the run configuration to use, e.g. input table or job name. install_folder: Custom workspace installation folder. Required if DQX is installed in a custom folder. - assume_user: Whether to assume a per-user installation when loading the run configuration (True as default, skipped if install_folder is provided). - product_name: Product/installation identifier used to resolve installation paths for config loading in install_folder is not provided ("dqx" as default). + assume_user: Whether to assume a per-user installation when loading the run configuration (use *True* if not provided, skipped if install_folder is provided). + product_name: Product/installation identifier used to resolve installation paths for config loading in install_folder is not provided (use "dqx" if not provided). """ installation = self.get_installation(assume_user, product_name, install_folder) return self._load_run_config(installation, run_config_name) @@ -92,8 +92,8 @@ def _load_run_config(installation: Installation, run_config_name: str | None) -> Load run configuration from the installation. Args: - installation: the installation object - run_config_name: name of the run configuration to use + installation: the installation object. + run_config_name: name of the run configuration to use, e.g. input table or job name. """ config = installation.load(WorkspaceConfig) return config.get_run_config(run_config_name) diff --git a/src/databricks/labs/dqx/engine.py b/src/databricks/labs/dqx/engine.py index 03b8cc99d..0aa7cc119 100644 --- a/src/databricks/labs/dqx/engine.py +++ b/src/databricks/labs/dqx/engine.py @@ -759,9 +759,9 @@ def save_results_in_table( quarantine_df: DataFrame with invalid rows to be saved (optional). output_config: Configuration describing where/how to write the valid rows. If omitted, falls back to the run config. quarantine_config: Configuration describing where/how to write the invalid rows (optional). If omitted, falls back to the run config. - run_config_name: Name of the run configuration to load when a config parameter is omitted. - product_name: Product/installation identifier used to resolve installation paths for config loading in install_folder is not provided ("dqx" as default). - assume_user: Whether to assume a per-user installation when loading the run configuration (True as default, skipped if install_folder is provided). + run_config_name: Name of the run configuration to load when a config parameter is omitted, e.g. input table or job name (use "default" if not provided). + product_name: Product/installation identifier used to resolve installation paths for config loading in install_folder is not provided (use "dqx" if not provided). + assume_user: Whether to assume a per-user installation when loading the run configuration (use *True* if not provided, skipped if install_folder is provided). install_folder: Custom workspace installation folder. Required if DQX is installed in a custom folder. Returns: diff --git a/src/databricks/labs/dqx/manager.py b/src/databricks/labs/dqx/manager.py index 4ce010cab..52ca572ac 100644 --- a/src/databricks/labs/dqx/manager.py +++ b/src/databricks/labs/dqx/manager.py @@ -1,8 +1,10 @@ +import logging from datetime import datetime from dataclasses import dataclass from functools import cached_property import pyspark.sql.functions as F +from pyspark.errors import AnalysisException from pyspark.sql import DataFrame, Column, SparkSession from databricks.labs.dqx.executor import DQCheckResult, DQRuleExecutorFactory @@ -10,6 +12,9 @@ DQRule, ) from databricks.labs.dqx.schema.dq_result_schema import dq_result_item_schema +from databricks.labs.dqx.utils import get_column_name_or_alias + +logger = logging.getLogger(__name__) @dataclass(frozen=True) @@ -53,18 +58,57 @@ def user_metadata(self) -> dict[str, str]: return self.engine_user_metadata or {} @cached_property - def filter_condition(self): + def filter_condition(self) -> Column: """ Returns the filter condition for the check. """ return F.expr(self.check.filter) if self.check.filter else F.lit(True) + @cached_property + def invalid_columns(self) -> list[str]: + """ + Returns list of invalid check columns in the input DataFrame. + """ + invalid_cols = [] + + if self.check.column is not None and self._is_invalid_column(self.check.column): + invalid_cols.append(get_column_name_or_alias(self.check.column)) + elif self.check.columns is not None: # either column or columns can be provided, but not both + for column in self.check.columns: + if self._is_invalid_column(column): + invalid_cols.append(get_column_name_or_alias(column)) + + return invalid_cols + + @cached_property + def has_invalid_columns(self) -> bool: + """ + Returns a boolean indicating whether any of the specified check columns are invalid in the input DataFrame. + """ + return bool(self.invalid_columns) + + @cached_property + def has_invalid_filter(self) -> bool: + """ + Returns a boolean indicating whether the filter is invalid in the input DataFrame. + """ + return self._is_invalid_column(self.filter_condition) + def process(self) -> DQCheckResult: """ - Process the data quality rule and return results as DQCheckResult containing: + Process the data quality rule (check) and return results as DQCheckResult containing: - Column with the check result - optional DataFrame with the results of the check + + Skip the check evaluation if column or columns, or filter in the check cannot be resolved in the input + DataFrame. Return the check result preserving all fields with message identifying invalid fields. """ + invalid_cols_message = self._get_invalid_cols_message() + if invalid_cols_message: + # overwrite message but preserve all other fields in the result + result_struct = self._build_result_struct(condition=F.lit(invalid_cols_message)) + return DQCheckResult(condition=result_struct, check_df=self.df) + executor = DQRuleExecutorFactory.create(self.check) raw_result = executor.apply(self.df, self.spark, self.ref_dfs) return self._wrap_result(raw_result) @@ -72,7 +116,6 @@ def process(self) -> DQCheckResult: def _wrap_result(self, raw_result: DQCheckResult) -> DQCheckResult: result_struct = self._build_result_struct(raw_result.condition) check_result = F.when(self.filter_condition & raw_result.condition.isNotNull(), result_struct) - return DQCheckResult(condition=check_result, check_df=raw_result.check_df) def _build_result_struct(self, condition: Column) -> Column: @@ -87,3 +130,44 @@ def _build_result_struct(self, condition: Column) -> Column: "user_metadata" ), ).cast(dq_result_item_schema) + + def _get_invalid_cols_message(self) -> str: + """ + Returns invalid columns message containing info about invalid columns to check should be applied to or filter. + """ + invalid_cols_message_parts = [] + + if self.has_invalid_columns: + logger.warning( + f"Skipping check evaluation '{self.check.name}' due to invalid check columns: {self.invalid_columns}" + ) + invalid_cols_message_parts.append( + f"Check evaluation skipped due to invalid check columns: {self.invalid_columns}" + ) + + if self.has_invalid_filter: + logger.warning(f"Skipping check '{self.check.name}' due to invalid check filter: '{self.check.filter}'") + invalid_cols_message_parts.append( + f"Check evaluation skipped due to invalid check filter: '{self.check.filter}'" + ) + + invalid_cols_message = "; ".join(invalid_cols_message_parts) + + return invalid_cols_message + + def _is_invalid_column(self, column: str | Column) -> bool: + """ + Returns True if the specified column is invalid (i.e., cannot be resolved in the input DataFrame), + otherwise False. + """ + try: + col_expr = F.expr(column) if isinstance(column, str) else column + _ = self.df.select(col_expr).schema # perform logical plan validation without triggering computation + except AnalysisException as e: + # if column is not accessible or column expression cannot be evaluated, an AnalysisException is thrown + logger.debug( + f"Invalid column '{column}' provided in the check '{self.check.name}'", + exc_info=e, + ) + return True + return False diff --git a/tests/integration/test_apply_checks.py b/tests/integration/test_apply_checks.py index 8ee1fe1f3..5c572800c 100644 --- a/tests/integration/test_apply_checks.py +++ b/tests/integration/test_apply_checks.py @@ -7831,3 +7831,290 @@ def test_apply_checks_and_save_in_tables_for_patterns_missing_quarantine_suffix( run_config_template=RunConfig(quarantine_config=OutputConfig("catalog.schema.table")), quarantine_table_suffix="", ) + + +def test_apply_checks_skip_checks_with_missing_columns(ws, spark): + dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS) + complex_cols_schema = ", arr_col array, map_col: map, struct_col: struct" + test_df = spark.createDataFrame([[1, 3, 3, [1], {"key1": 1}, {"field1": 1}]], SCHEMA + complex_cols_schema) + + checks = [ + # pass - no issues + DQRowRule( + name="a_is_null_or_empty", criticality="warn", check_func=check_funcs.is_not_null_and_not_empty, column="a" + ), + DQRowRule( + criticality="warn", + check_func=check_funcs.is_not_null, + column=F.try_element_at("arr_col", F.lit(1)), + ), + DQRowRule( + criticality="warn", + check_func=check_funcs.is_not_null, + column=F.try_element_at("map_col", F.lit("key1")), + ), + DQRowRule( + criticality="warn", + check_func=check_funcs.is_not_null, + column="struct_col.field1", + ), + # invalid filter + DQRowRule( + name="b_is_null_or_empty", + criticality="error", + check_func=check_funcs.is_not_null_and_not_empty, + column=F.col("b"), + filter="missing_col > 0", + ), + # invalid column + DQRowRule( + criticality="warn", + check_func=check_funcs.is_not_null_and_not_empty, + column=F.col("missing_col"), + filter="a > 0", + user_metadata={"tag1": "value1", "tag2": "value2"}, + ), + # invalid column in for each rule + *DQForEachColRule( + check_func=check_funcs.is_not_null, + columns=["missing_col"], + criticality="error", + ).get_rules(), + # invalid columns + DQRowRule( + name="missing_col_sql_expression", + criticality="error", + check_func=check_funcs.sql_expression, + check_func_kwargs={ + "expression": "missing_col > 0", + "msg": "missing_col is less than 0", + }, + columns=["missing_col"], + ), + # invalid columns and filter + DQDatasetRule( + name="missing_col_is_unique", + criticality="error", + check_func=check_funcs.is_unique, + columns=["missing_col"], + filter="missing_col > 0", + ), + ] + + checked = dq_engine.apply_checks(test_df, checks) + + expected = spark.createDataFrame( + [ + [ + 1, + 3, + 3, + [1], + {"key1": 1}, + {"field1": 1}, + [ + { + "name": "b_is_null_or_empty", + "message": "Check evaluation skipped due to invalid check filter: 'missing_col > 0'", + "columns": ["b"], + "filter": "missing_col > 0", + "function": "is_not_null_and_not_empty", + "run_time": RUN_TIME, + "user_metadata": {}, + }, + { + "name": "missing_col_is_null", + "message": "Check evaluation skipped due to invalid check columns: ['missing_col']", + "columns": ["missing_col"], + "filter": None, + "function": "is_not_null", + "run_time": RUN_TIME, + "user_metadata": {}, + }, + { + "name": "missing_col_sql_expression", + "message": "Check evaluation skipped due to invalid check columns: ['missing_col']", + "columns": ["missing_col"], + "filter": None, + "function": "sql_expression", + "run_time": RUN_TIME, + "user_metadata": {}, + }, + { + "name": "missing_col_is_unique", + "message": "Check evaluation skipped due to invalid check columns: ['missing_col']; " + "Check evaluation skipped due to invalid check filter: 'missing_col > 0'", + "columns": ["missing_col"], + "filter": "missing_col > 0", + "function": "is_unique", + "run_time": RUN_TIME, + "user_metadata": {}, + }, + ], + [ + { + "name": "missing_col_is_null_or_empty", + "message": "Check evaluation skipped due to invalid check columns: ['missing_col']", + "columns": ["missing_col"], + "filter": "a > 0", + "function": "is_not_null_and_not_empty", + "run_time": RUN_TIME, + "user_metadata": {"tag1": "value1", "tag2": "value2"}, + }, + ], + ] + ], + SCHEMA + complex_cols_schema + REPORTING_COLUMNS, + ) + assert_df_equality(checked, expected, ignore_nullable=True) + + +def test_apply_checks_by_metadata_skip_checks_with_missing_columns(ws, spark): + dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS) + complex_cols_schema = ", arr_col array, map_col: map, struct_col: struct" + test_df = spark.createDataFrame([[1, 3, 3, [1], {"key1": 1}, {"field1": 1}]], SCHEMA + complex_cols_schema) + + checks = [ + { + "criticality": "warn", + "check": { + "function": "is_not_null_and_not_empty", + "arguments": {"column": "a"}, + }, + }, + { + "criticality": "warn", + "check": { + "function": "is_not_null", + "arguments": {"column": "try_element_at(arr_col, 1)"}, + }, + }, + { + "criticality": "warn", + "check": { + "function": "is_not_null", + "arguments": {"column": "try_element_at(map_col, 'key1')"}, + }, + }, + { + "criticality": "warn", + "check": { + "function": "is_not_null", + "arguments": {"column": "struct_col.field1"}, + }, + }, + { + "criticality": "error", + "filter": "missing_col > 0", + "check": { + "function": "is_not_null_and_not_empty", + "arguments": {"column": "b"}, + }, + }, + { + "criticality": "warn", + "filter": "a > 0", + "check": { + "function": "is_not_null_and_not_empty", + "arguments": { + "column": "missing_col", + }, + }, + "user_metadata": {"tag1": "value1", "tag2": "value2"}, + }, + { + "criticality": "error", + "check": { + "function": "is_not_null", + "for_each_column": ["missing_col"], + }, + }, + { + "name": "missing_col_sql_expression", + "criticality": "error", + "check": { + "function": "sql_expression", + "arguments": { + "expression": "missing_col > 0", + "msg": "missing_col is less than 0", + "columns": ["missing_col"], + }, + }, + }, + { + "name": "missing_col_is_unique", + "criticality": "error", + "filter": "missing_col > 0", + "check": { + "function": "is_unique", + "arguments": {"columns": ["missing_col"]}, + }, + }, + ] + + checked = dq_engine.apply_checks_by_metadata(test_df, checks) + + expected = spark.createDataFrame( + [ + [ + 1, + 3, + 3, + [1], + {"key1": 1}, + {"field1": 1}, + [ + { + "name": "b_is_null_or_empty", + "message": "Check evaluation skipped due to invalid check filter: 'missing_col > 0'", + "columns": ["b"], + "filter": "missing_col > 0", + "function": "is_not_null_and_not_empty", + "run_time": RUN_TIME, + "user_metadata": {}, + }, + { + "name": "missing_col_is_null", + "message": "Check evaluation skipped due to invalid check columns: ['missing_col']", + "columns": ["missing_col"], + "filter": None, + "function": "is_not_null", + "run_time": RUN_TIME, + "user_metadata": {}, + }, + { + "name": "missing_col_sql_expression", + "message": "Check evaluation skipped due to invalid check columns: ['missing_col']", + "columns": ["missing_col"], + "filter": None, + "function": "sql_expression", + "run_time": RUN_TIME, + "user_metadata": {}, + }, + { + "name": "missing_col_is_unique", + "message": "Check evaluation skipped due to invalid check columns: ['missing_col']; " + "Check evaluation skipped due to invalid check filter: 'missing_col > 0'", + "columns": ["missing_col"], + "filter": "missing_col > 0", + "function": "is_unique", + "run_time": RUN_TIME, + "user_metadata": {}, + }, + ], + [ + { + "name": "missing_col_is_null_or_empty", + "message": "Check evaluation skipped due to invalid check columns: ['missing_col']", + "columns": ["missing_col"], + "filter": "a > 0", + "function": "is_not_null_and_not_empty", + "run_time": RUN_TIME, + "user_metadata": {"tag1": "value1", "tag2": "value2"}, + }, + ], + ] + ], + SCHEMA + complex_cols_schema + REPORTING_COLUMNS, + ) + assert_df_equality(checked, expected, ignore_nullable=True)