- })
-
+## How DQX works
-Invalid data can be quarantined to make sure bad data is never written to the output.
+### Option 1: Apply checks and quarantine "bad" data.
+
+Apply checks on the DataFrame and quarantine invalid records to ensure "bad" data is never written to the output.
-})
+
+
-In the Lakehouse architecture, the validation of new data should happen at the time of data entry into the Curated Layer
-to make sure bad data is not propagated to the subsequent layers. With DQX you can easily quarantine invalid data and re-ingest it
-after curation to ensure that data quality constraints are met.
+### DQX usage in the Lakehouse Architecture
+In the Lakehouse architecture, new data validation should happen during data entry into the curated layer to ensure bad data is not propagated to the subsequent layers.
+With DQX, you can quickly quarantine invalid data and re-ingest it after curation to ensure data quality constraints are met.
+The data quality can be monitored in real-time between layers, and the quarantine process can be automated.
+})
@@ -38,8 +46,8 @@ after curation to ensure that data quality constraints are met.
## When to use DQX
-* Use DQX if you need pro-active monitoring (before data is written to a target table).
+* Use DQX if you need pro-active monitoring (before data is written to a target table), especially in the streaming pipelines.
* For monitoring data quality of already persisted data in Delta tables (post-factum monitoring), try
[Databricks Lakehouse Monitoring](https://docs.databricks.com/en/lakehouse-monitoring/index.html).
-* DQX can be integrated with DLT for data quality checking but your first choice for DLT pipelines should be [DLT Expectations](https://docs.databricks.com/en/delta-live-tables/expectations.html#what-are-delta-live-tables-expectations). DQX can be used to profile data and generate DLT expectation candidates.
+* DQX can be integrated with DLT to check data quality. Still, your first choice for DLT pipelines should be [DLT Expectations](https://docs.databricks.com/en/delta-live-tables/expectations.html#what-are-delta-live-tables-expectations). DQX can be used to profile data and generate DLT expectation candidates.
* DQX can be integrated with other data transformation frameworks that support PySpark, such as [dbt](https://docs.getdbt.com/docs/build/python-models). However, this integration is limited to dbt Python models and does not extend to dbt SQL models.
diff --git a/docs/dqx/docs/reference/engine.mdx b/docs/dqx/docs/reference/engine.mdx
index 95124689..829de3f1 100644
--- a/docs/dqx/docs/reference/engine.mdx
+++ b/docs/dqx/docs/reference/engine.mdx
@@ -1,34 +1,45 @@
+import Admonition from '@theme/Admonition';
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
# DQX Engine
-To perform data quality checking with DQX, you need to create `DQEngine` object.
+To perform data quality checking with DQX, you must create a `DQEngine` object.
The engine requires a Databricks [workspace client](https://docs.databricks.com/aws/en/dev-tools/sdk-python) for authentication and interaction with the Databricks workspace.
-When running the code on a Databricks workspace, the workspace client is automatically authenticated, whether DQX is used in a notebook, script, or as part of a job/workflow.
+When running the code on a Databricks workspace, the workspace client is automatically authenticated, whether DQX is used in a notebook, script, or job/workflow.
You only need the following code to create the workspace client if you run DQX on Databricks workspace:
-```python
-from databricks.sdk import WorkspaceClient
-from databricks.labs.dqx.engine import DQEngine
-ws = WorkspaceClient()
-dq_engine = DQEngine(ws)
-```
+
+
diff --git a/docs/dqx/docs/reference/quality_rules.mdx b/docs/dqx/docs/reference/quality_rules.mdx
index 25ca722d..d0420473 100644
--- a/docs/dqx/docs/reference/quality_rules.mdx
+++ b/docs/dqx/docs/reference/quality_rules.mdx
@@ -1,52 +1,491 @@
+import Admonition from '@theme/Admonition';
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
# Quality rules
-This page provides a reference for the quality rule functions (checks) available in DQX.
-
-## Quality rule functions (checks)
-
-The following quality rules / functions are currently available:
-
-| Check | Description | Arguments |
-| -------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| is_not_null | Check if input column is not null | col_name: column name to check |
-| is_not_empty | Check if input column is not empty | col_name: column name to check |
-| is_not_null_and_not_empty | Check if input column is not null or empty | col_name: column name to check; trim_strings: boolean flag to trim spaces from strings |
-| value_is_in_list | Check if the provided value is present in the input column. | col_name: column name to check; allowed: list of allowed values |
-| value_is_not_null_and_is_in_list | Check if provided value is present if the input column is not null | col_name: column name to check; allowed: list of allowed values |
-| is_not_null_and_not_empty_array | Check if input array column is not null or empty | col_name: column name to check |
-| is_in_range | Check if input column is in the provided range (inclusive of both boundaries) | col_name: column name to check; min_limit: min limit value; max_limit: max limit value; min_limit_col_expr: min limit column name or expr; max_limit_col_expr: max limit column name or expr |
-| is_not_in_range | Check if input column is not within defined range (inclusive of both boundaries) | col_name: column name to check; min_limit: min limit value; max_limit: max limit value; min_limit_col_expr: min limit column name or expr; max_limit_col_expr: max limit column name or expr |
-| not_less_than | Check if input column is not less than the provided limit | col_name: column name to check; limit: limit value |
-| not_greater_than | Check if input column is not greater than the provided limit | col_name: column name to check; limit: limit value |
-| is_valid_date | Check if input column is a valid date | col_name: column name to check; date_format: date format (e.g. 'yyyy-mm-dd') |
-| is_valid_timestamp | Check if input column is a valid timestamp | col_name: column name to check; timestamp_format: timestamp format (e.g. 'yyyy-mm-dd HH:mm:ss') |
-| not_in_future | Check if input column defined as date is not in the future (future defined as current_timestamp + offset) | col_name: column name to check; offset: offset to use; curr_timestamp: current timestamp, if not provided current_timestamp() function is used |
-| not_in_near_future | Check if input column defined as date is not in the near future (near future defined as grater than current timestamp but less than current timestamp + offset) | col_name: column name to check; offset: offset to use; curr_timestamp: current timestamp, if not provided current_timestamp() function is used |
-| is_older_than_n_days | Check if input column is older than n number of days | col_name: column name to check; days: number of days; curr_date: current date, if not provided current_date() function is used |
-| is_older_than_col2_for_n_days | Check if one column is not older than another column by n number of days | col_name1: first column name to check; col_name2: second column name to check; days: number of days |
-| regex_match | Check if input column matches a given regex | col_name: column name to check; regex: regex to check; negate: if the condition should be negated (true) or not |
-| sql_expression | Check if input column is matches the provided sql expression, eg. a = 'str1', a > b | expression: sql expression to check; msg: optional message to output; name: optional name of the resulting column; negate: if the condition should be negated |
-
-You can check implementation details of the rules [here](https://github.com/databrickslabs/dqx/blob/main/src/databricks/labs/dqx/col_functions.py).
-
-## Apply filters on checks
+This page provides a reference for the quality checks (rule functions) available in DQX.
+
+## Row-level quality checks
+
+The following row-level checks are currently available in DQX.
+These checks are applied to each row of a PySpark DataFrame and generate issue reports as additional columns.
+You can also define your own custom checks (see [Creating custom checks](#creating-custom-checks)).
+
+**Available DQX engine methods**
| Check | Description | Arguments | | ---------------------------------- | --------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| apply_checks | Applies quality checks to the DataFrame and returns a DataFrame with reporting columns. | df: DataFrame to check; checks: List of checks to the DataFrame. Each check is an instance of DQRule class. | -| apply_checks_and_split | Applies quality checks to the DataFrame and returns valid and invalid (quarantine) DataFrames with reporting columns. | df: DataFrame to check; checks: List of checks to apply to the DataFrame. Each check is an instance of DQRule class. | +| apply_checks | Applies quality checks to the DataFrame and returns a DataFrame with reporting columns. | df: DataFrame to check; checks: List of checks to the DataFrame. Each check is an instance of the DQRule class. | +| apply_checks_and_split | Applies quality checks to the DataFrame and returns valid and invalid (quarantine) DataFrames with reporting columns. | df: DataFrame to check; checks: List of checks to apply to the DataFrame. Each check is an instance of the DQRule class. | | apply_checks_by_metadata | Applies quality checks defined as a dictionary to the DataFrame and returns a DataFrame with reporting columns. | df: DataFrame to check. checks: List of dictionaries describing checks; custom_check_functions: Optional dictionary with custom check functions (e.g., globals() of the calling module). | | 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 dictionaries describing checks; custom_check_functions: Optional dictionary with custom check functions (e.g., globals() of the calling module). | | 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. | @@ -41,3 +52,4 @@ The following table outlines the available methods of the DQEngine and their fun | save_checks_in_workspace_file | Saves checks to a file (YAML) in the Databricks workspace. | checks: List of checks to save; workspace_path: Destination path for the checks file in the workspace. | | save_checks_in_installation | Saves checks to the installation folder as a YAML file. | checks: List of checks to save; run_config_name: Name of the run config to use; assume_user: If True, assume user installation. | | load_run_config | Loads run configuration from the installation folder. | run_config_name: Name of the run config to use; assume_user: If True, assume user installation. | +
+
+
+You can explore the implementation details of the rules [here](https://github.com/databrickslabs/dqx/blob/main/src/databricks/labs/dqx/col_functions.py).
+If you have a custom check that could be broadly useful, feel free to submit a PR to [DQX](https://github.com/databrickslabs/dqx) (see the [contribution guide](/docs/dev/contributing) for details).
+
+### Usage examples of row-level checks
+
+Below are fully specified examples of how to use each check in YAML format and with DQX classes. Both are equivalent and can be used interchangeably.
+
+The `criticality` field can be either "error" (data goes only into the 'bad' or 'quarantine' DataFrame) or "warn" (data goes into both DataFrames).
+For brevity, the `name` field in the examples is omitted, meaning it will be auto-generated in the results.
+
+**Available row-level checks**
+| Check | Description | Arguments | +| -------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| is_not_null | Checks whether the values in the input column are not null. | col_name: column name to check | +| is_not_empty | Checks whether the values in the input column are not empty (but may be null). | col_name: column name to check | +| is_not_null_and_not_empty | Checks whether the values in the input column are not null and not empty. | col_name: column name to check; 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). | col_name: column name to check; 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. | col_name: column name to check; 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. | col_name: column name to check | +| is_in_range | Checks whether the values in the input column are in the provided range (inclusive of both boundaries). | col_name: column name to check; min_limit: min limit as number, date, timestamp, column name or expression; max_limit: max limit as number, date, timestamp, column name or expression | +| is_not_in_range | Checks whether the values in the input column are outside the provided range (inclusive of both boundaries). | col_name: column name to check; min_limit: min limit as number, date, timestamp, column name or expression; max_limit: max limit as number, date, timestamp, column name or expression | +| is_not_less_than | Checks whether the values in the input column are not less than the provided limit. | col_name: column name to check; limit: limit as number, date, timestamp, column name or expression | +| is_not_greater_than | Checks whether the values in the input column are not greater than the provided limit. | col_name: column name to check; limit: limit as number, date, timestamp, column name or expression | +| is_valid_date | Checks whether the values in the input column have valid date formats. | col_name: column name to check; 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. | col_name: column name to check; 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). | col_name: column name to check; 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). | col_name: column name to check; 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. | col_name: column name to check; days: number of days; curr_date: current date, if not provided current_date() function is used | +| 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. | col_name1: first column name to check; col_name2: second column name to check; days: number of days | +| is_unique | Checks whether the values in the input column are unique and reports an issue for each row that contains a duplicate value. Null values are not considered duplicates, following the ANSI SQL standard. | col_name: column name to check: window_spec: optional window specification as a string or column object, you must handle NULLs correctly using coalesce() to prevent rows exclusion | +| regex_match | Checks whether the values in the input column match a given regex. | col_name: column name to check; regex: regex to check; negate: if the condition should be negated (true) or not | +| sql_expression | Checks whether the values meet the condition provided as an SQL expression, e.g. a = 'str1', a > b | expression: sql expression to check; msg: optional message to output; name: optional name of the resulting column; negate: if the condition should be negated | +
+
+
+**Checks define in YAML**
+```yaml +# is_not_null check +- criticality: error + check: + function: is_not_null + arguments: + col_name: col1 + +# is_not_empty check +- criticality: error + check: + function: is_not_empty + arguments: + col_name: col1 + +# is_not_null_and_not_empty check +- criticality: error + check: + function: is_not_null_and_not_empty + arguments: + col_name: col1 + trim_strings: true + +# is_in_list check +- criticality: error + check: + function: is_in_list + arguments: + col_name: col2 + allowed: + - 1 + - 2 + - 3 + +# is_not_null_and_is_in_list check +- criticality: error + check: + function: is_not_null_and_is_in_list + arguments: + col_name: col2 + allowed: + - 1 + - 2 + - 3 + +# is_not_null_and_not_empty_array check +- criticality: error + check: + function: is_not_null_and_not_empty_array + arguments: + col_name: col4 + +# is_in_range check +- criticality: error + check: + function: is_in_range + arguments: + col_name: col2 + min_limit: 1 + max_limit: 10 +- criticality: error + check: + function: is_in_range + arguments: + col_name: col5 + min_limit: 2025-01-01 + max_limit: 2025-02-24 +- criticality: error + check: + function: is_in_range + arguments: + col_name: col6 + min_limit: 2025-01-01 00:00:00 + max_limit: 2025-02-24 01:00:00 +- criticality: error + check: + function: is_in_range + arguments: + col_name: col3 + min_limit: col2 + max_limit: col2 * 2 + +# is_not_in_range check +- criticality: error + check: + function: is_not_in_range + arguments: + col_name: col2 + min_limit: 11 + max_limit: 20 +- criticality: error + check: + function: is_not_in_range + arguments: + col_name: col5 + min_limit: 2025-02-25 + max_limit: 2025-02-26 +- criticality: error + check: + function: is_not_in_range + arguments: + col_name: col6 + min_limit: 2025-02-25 00:00:00 + max_limit: 2025-02-26 01:00:00 +- criticality: error + check: + function: is_not_in_range + arguments: + col_name: col3 + min_limit: col2 + 10 + max_limit: col2 * 10 + +# is_not_less_than check +- criticality: error + check: + function: is_not_less_than + arguments: + col_name: col2 + limit: 0 +- criticality: error + check: + function: is_not_less_than + arguments: + col_name: col5 + limit: 2025-01-01 +- criticality: error + check: + function: is_not_less_than + arguments: + col_name: col6 + limit: 2025-01-01 01:00:00 +- criticality: error + check: + function: is_not_less_than + arguments: + col_name: col3 + limit: col2 - 10 + +# is_not_greater_than check +- criticality: error + check: + function: is_not_greater_than + arguments: + col_name: col2 + limit: 10 +- criticality: error + check: + function: is_not_greater_than + arguments: + col_name: col5 + limit: 2025-03-01 +- criticality: error + check: + function: is_not_greater_than + arguments: + col_name: col6 + limit: 2025-03-24 01:00:00 +- criticality: error + check: + function: is_not_greater_than + arguments: + col_name: col3 + limit: col2 + 10 + +# is_valid_date check +- criticality: error + check: + function: is_valid_date + arguments: + col_name: col5 +- criticality: error + name: col5_is_not_valid_date2 + check: + function: is_valid_date + arguments: + col_name: col5 + date_format: yyyy-MM-dd + +# is_valid_timestamp check +- criticality: error + check: + function: is_valid_timestamp + arguments: + col_name: col6 + timestamp_format: yyyy-MM-dd HH:mm:ss +- criticality: error + name: col6_is_not_valid_timestamp2 + check: + function: is_valid_timestamp + arguments: + col_name: col6 + +# is_not_in_future check +- criticality: error + check: + function: is_not_in_future + arguments: + col_name: col6 + offset: 86400 + +# is_not_in_near_future check +- criticality: error + check: + function: is_not_in_near_future + arguments: + col_name: col6 + offset: 36400 + +# is_older_than_n_days check +- criticality: error + check: + function: is_older_than_n_days + arguments: + col_name: col5 + days: 10000 + +# is_older_than_col2_for_n_days check +- criticality: error + check: + function: is_older_than_col2_for_n_days + arguments: + col_name1: col5 + col_name2: col6 + days: 2 + +# is_unique check +- criticality: error + check: + function: is_unique + arguments: + col_name: col1 + +# is_unique check with custom window +# default value for NULL in the time column of the window spec must be provided using coalesce() to prevent rows exclusion! +- criticality: error + name: col1_is_not_unique2 + check: + function: is_unique + arguments: + col_name: col1 + window_spec: window(coalesce(col6, '1970-01-01'), '10 minutes') + +# regex_match check +- criticality: error + check: + function: regex_match + arguments: + col_name: col2 + regex: '[0-9]+' + negate: false + +# sql_expression check +- criticality: error + check: + function: sql_expression + arguments: + expression: col3 > col2 and col3 < 10 + msg: col3 is greater than col2 and col3 less than 10 + name: custom_output_name + negate: false +``` +
+
+
+## Applying filters on checks
You can apply checks to a part of the DataFrame by using a `filter`.
-For example, to ensure that a column `a` is not null only when a column `b` is positive, you can define the check as follows:
+For example, to check that a column `a` is not null only when a column `b` is positive, you can define the check as follows:
```yaml
- criticality: error
- filter: b > 0
+ filter: col2 > 0
check:
function: is_not_null
arguments:
- col_name: a
+ col_name: col1
```
-## Creating your own checks
+## Creating custom checks
-### Use sql expression
+### Custom checks with SQL Expression
-If a check that you need does not exist in DQX, you can define them using sql expression rule (`sql_expression`),
+You can define custom checks using SQL Expression rule (`sql_expression`),
for example:
```yaml
- criticality: error
@@ -57,83 +496,96 @@ for example:
msg: col1 ends with 'foo'
```
-Sql expression is also useful if you want to make cross-column validation, for example:
+SQL Expressions are also useful if you need to make cross-column validation, for example:
```yaml
- criticality: error
check:
function: sql_expression
arguments:
- expression: a > b
- msg: a is greater than b
+ expression: col1 > col2
+ msg: col1 is greater than col2
```
-### Define custom check functions
+### Custom checks as a Python function
-If you need a reusable check or want to implement more complex logic, you can define your own custom check functions.
-A check function is a callable that returns a `pyspark.sql.Column`. For example:
+If you need a reusable check or want to implement more complex logic which is challenging to implement with SQL, you can define your own custom check functions.
+A check function is a callable that returns a `pyspark.sql.Column`.
-```python
-import pyspark.sql.functions as F
-from pyspark.sql import Column
-from databricks.labs.dqx.col_functions import make_condition
+#### Custom check example
-def ends_with_foo(col_name: str) -> Column:
- column = F.col(col_name)
- return make_condition(column.endswith("foo"), f"Column {col_name} ends with foo", f"{col_name}_ends_with_foo")
-```
+**Checks defined using DQX classes**
+```python +from databricks.labs.dqx.col_functions import * +from datetime import datetime + +checks = [ + DQRule( + criticality="error", + check=is_not_null("col1") + ), + DQRule( + criticality="error", + check=is_not_empty("col1") + ), + DQRule( + criticality="error", + check=is_not_null_and_not_empty("col1", trim_strings=True) + ), + DQRule( + criticality="error", + check=is_in_list("col2", [1, 2, 3]) + ), + DQRule( + criticality="error", + check=is_not_null_and_is_in_list("col2", [1, 2, 3]) + ), + DQRule( + criticality="error", + check=is_not_null_and_not_empty_array("col4") + ), + DQRule( + criticality="error", + check=is_in_range("col2", min_limit=1, max_limit=10) + ), + DQRule( + criticality="error", + check=is_in_range("col5", min_limit=datetime(2025, 1, 1).date(), max_limit=datetime(2025, 2, 24).date()) + ), + DQRule( + criticality="error", + check=is_in_range("col6", min_limit=datetime(2025, 1, 1, 0, 0, 0), max_limit=datetime(2025, 2, 24, 1, 0, 0)) + ), + DQRule( + criticality="error", + check=is_in_range("col3", min_limit="col2", max_limit="col2 * 2") + ), + DQRule( + criticality="error", + check=is_not_in_range("col2", min_limit=11, max_limit=20) + ), + DQRule( + criticality="error", + check=is_not_in_range("col5", min_limit=datetime(2025, 2, 25).date(), max_limit=datetime(2025, 2, 26).date()) + ), + DQRule( + criticality="error", + check=is_not_in_range("col6", min_limit=datetime(2025, 2, 25, 0, 0, 0), max_limit=datetime(2025, 2, 26, 1, 0, 0)) + ), + DQRule( + criticality="error", + check=is_not_in_range("col3", min_limit="col2 + 10", max_limit="col2 * 10") + ), + DQRule( + criticality="error", + check=is_not_less_than("col2", limit=0) + ), + DQRule( + criticality="error", + check=is_not_less_than("col5", limit=datetime(2025, 1, 1).date()) + ), + DQRule( + criticality="error", + check=is_not_less_than("col6", limit=datetime(2025, 1, 1, 1, 0, 0)) + ), + DQRule( + criticality="error", + check=is_not_less_than("col3", limit="col2 - 10") + ), + DQRule( + criticality="error", + check=is_not_greater_than("col2", limit=10) + ), + DQRule( + criticality="error", + check=is_not_greater_than("col5", limit=datetime(2025, 3, 1).date()) + ), + DQRule( + criticality="error", + check=is_not_greater_than("col6", limit=datetime(2025, 3, 24, 1, 0, 0)) + ), + DQRule( + criticality="error", + check=is_not_greater_than("col3", limit="col2 + 10") + ), + DQRule( + criticality="error", + check=is_valid_date("col5") + ), + DQRule( + criticality="error", + check=is_valid_date("col5", date_format="yyyy-MM-dd"), + name="col5_is_not_valid_date2" + ), + DQRule( + criticality="error", + check=is_valid_timestamp("col6") + ), + DQRule( + criticality="error", + check=is_valid_timestamp("col6", timestamp_format="yyyy-MM-dd HH:mm:ss"), + name="col6_is_not_valid_timestamp2" + ), + DQRule( + criticality="error", + check=is_not_in_future("col6", offset=86400) + ), + DQRule( + criticality="error", + check=is_not_in_near_future("col6", offset=36400) + ), + DQRule( + criticality="error", + check=is_older_than_n_days("col5", days=10000) + ), + DQRule( + criticality="error", + check=is_older_than_col2_for_n_days("col5", "col6", days=2) + ), + DQRule( + criticality="error", + check=is_unique("col1") + ), + DQRule( + criticality="error", + name="col1_is_not_unique2", + # default value for NULL in the time column of the window spec must be provided using coalesce() to prevent rows exclusion! + check=is_unique("col1", window_spec=F.window(F.coalesce(F.col("col6"), F.lit(datetime(1970, 1, 1))), "10 minutes")) + ), + DQRule( + criticality="error", + check=regex_match("col2", regex="[0-9]+", negate=False) + ), + DQRule( + criticality="error", + check=sql_expression( + expression="col3 > col2 and col3 < 10", + msg="col3 is greater than col2 and col3 less than 10", + name="custom_output_name", + negate=False + ) + ), +] +``` +- Improve your data quality now 🚀 + Improve your Data Quality now 🚀
Follow our comprehensive guide to get up and running with DQX in no time. @@ -33,11 +33,11 @@ const Capabilities = () => { }, { title: 'Data Format Agnostic', - description: 'Works seamlessly with Spark DataFrames.', + description: 'Works seamlessly with PySpark DataFrames.', icon: FileText, }, { - title: 'Spark Batch & Streaming Support', + title: 'Spark Batch & Spark Structured Streaming Support', description: 'Includes Delta Live Tables (DLT) integration.', icon: Activity, }, @@ -57,8 +57,8 @@ const Capabilities = () => { icon: Grid, }, { - title: 'Profiling & Rule Generation', - description: 'Automatically profile and generate data quality rule candidates.', + title: 'Profiling & Quality Rules Generation', + description: 'Automatically profile input data and generate data quality rule candidates.', icon: BarChart2, }, { @@ -67,7 +67,7 @@ const Capabilities = () => { icon: Code, }, { - title: 'Validation Summary & Dashboard', + title: 'Validation Summary & Quality Dashboard', description: 'Track and identify data quality issues effectively.', icon: PieChart, }, @@ -116,7 +116,7 @@ const Hero = () => {
DQX is a data quality framework for Apache Spark that enables you to define, monitor, and - react to data quality issues in your data pipelines. + address data quality issues in your Python-based data pipelines.
{/* Call to Action Buttons */} diff --git a/src/databricks/labs/dqx/col_functions.py b/src/databricks/labs/dqx/col_functions.py index 43f18c6f..f8dbfdfd 100644 --- a/src/databricks/labs/dqx/col_functions.py +++ b/src/databricks/labs/dqx/col_functions.py @@ -3,6 +3,7 @@ import pyspark.sql.functions as F from pyspark.sql import Column +from pyspark.sql.window import Window def make_condition(condition: Column, message: Column | str, alias: str) -> Column: @@ -22,13 +23,8 @@ def make_condition(condition: Column, message: Column | str, alias: str) -> Colu return (F.when(condition, msg_col).otherwise(F.lit(None).cast("string"))).alias(_cleanup_alias_name(alias)) -def _cleanup_alias_name(col_name: str) -> str: - # avoid issues with structs - return col_name.replace(".", "_") - - -def is_not_null_and_not_empty(col_name: str, trim_strings: bool = False) -> Column: - """Creates a condition column to check if value is null or empty. +def is_not_null_and_not_empty(col_name: str, trim_strings: bool | None = False) -> Column: + """Checks whether the values in the input column are not null and not empty. :param col_name: column name to check :param trim_strings: boolean flag to trim spaces from strings @@ -42,7 +38,7 @@ def is_not_null_and_not_empty(col_name: str, trim_strings: bool = False) -> Colu def is_not_empty(col_name: str) -> Column: - """Creates a condition column to check if value is empty (but could be null). + """Checks whether the values in the input column are not empty (but may be null). :param col_name: column name to check :return: Column object for condition @@ -53,7 +49,7 @@ def is_not_empty(col_name: str) -> Column: def is_not_null(col_name: str) -> Column: - """Creates a condition column to check if value is null. + """Checks whether the values in the input column are not null. :param col_name: column name to check :return: Column object for condition @@ -62,13 +58,16 @@ def is_not_null(col_name: str) -> Column: return make_condition(column.isNull(), f"Column {col_name} is null", f"{col_name}_is_null") -def value_is_not_null_and_is_in_list(col_name: str, allowed: list) -> Column: - """Creates a condition column to check if value is null or not in the list of allowed values. +def is_not_null_and_is_in_list(col_name: str, allowed: list) -> Column: + """Checks whether the values in the input column are not null and present in the list of allowed values. :param col_name: column name to check :param allowed: list of allowed values (actual values or Column objects) :return: Column object for condition """ + if not allowed: + raise ValueError("allowed list is not provided.") + allowed_cols = [item if isinstance(item, Column) else F.lit(item) for item in allowed] column = F.col(col_name) condition = column.isNull() | ~column.isin(*allowed_cols) @@ -78,21 +77,25 @@ def value_is_not_null_and_is_in_list(col_name: str, allowed: list) -> Column: "", F.lit("Value "), F.when(column.isNull(), F.lit("null")).otherwise(column.cast("string")), - F.lit(" is not in the allowed list: ["), + F.lit(" is null or not in the allowed list: ["), F.concat_ws(", ", *allowed_cols), F.lit("]"), ), - f"{col_name}_value_is_not_in_the_list", + f"{col_name}_is_null_or_is_not_in_the_list", ) -def value_is_in_list(col_name: str, allowed: list) -> Column: - """Creates a condition column to check if value not in the list of allowed values (could be null). +def is_in_list(col_name: str, allowed: list) -> Column: + """Checks whether the values in the input column are present in the list of allowed values + (null values are allowed). :param col_name: column name to check :param allowed: list of allowed values (actual values or Column objects) :return: Column object for condition """ + if not allowed: + raise ValueError("allowed list is not provided.") + allowed_cols = [item if isinstance(item, Column) else F.lit(item) for item in allowed] column = F.col(col_name) condition = ~column.isin(*allowed_cols) @@ -106,7 +109,7 @@ def value_is_in_list(col_name: str, allowed: list) -> Column: F.concat_ws(", ", *allowed_cols), F.lit("]"), ), - f"{col_name}_value_is_not_in_the_list", + f"{col_name}_is_not_in_the_list", ) @@ -114,7 +117,7 @@ def value_is_in_list(col_name: str, allowed: list) -> Column: def sql_expression(expression: str, msg: str | None = None, name: str | None = None, negate: bool = False) -> Column: - """Creates a condition column from the SQL expression. + """Checks whether the condition provided as an SQL expression is met. :param expression: SQL expression :param msg: optional message of the `Column` type, automatically generated if None @@ -137,8 +140,8 @@ def sql_expression(expression: str, msg: str | None = None, name: str | None = N return make_condition(expr_col, F.concat_ws("", F.lit(f"Value matches expression: {expression_msg}")), name) -def is_older_than_col2_for_n_days(col_name1: str, col_name2: str, days: int) -> Column: - """Creates a condition column for case when one date or timestamp column is older than another column by N days. +def is_older_than_col2_for_n_days(col_name1: str, col_name2: str, days: int = 0) -> Column: + """Checks whether the values in one input column are at least N days older than the values in another column. :param col_name1: first column :param col_name2: second column @@ -159,13 +162,12 @@ def is_older_than_col2_for_n_days(col_name1: str, col_name2: str, days: int) -> col2_date, F.lit(f"' for more than {days} days"), ), - f"is_col_{col_name1}_older_than_{col_name2}_for_N_days", + f"is_col_{col_name1}_older_than_{col_name2}_for_n_days", ) def is_older_than_n_days(col_name: str, days: int, curr_date: Column | None = None) -> Column: - """Creates a condition column for case when specified date or timestamp column is older (compared to current date) - than N days. + """Checks whether the values in the input column are at least N days older than the current date. :param col_name: name of the column to check :param days: number of days @@ -188,13 +190,13 @@ def is_older_than_n_days(col_name: str, days: int, curr_date: Column | None = No curr_date, F.lit(f"' for more than {days} days"), ), - f"is_col_{col_name}_older_than_N_days", + f"is_col_{col_name}_older_than_n_days", ) -def not_in_future(col_name: str, offset: int = 0, curr_timestamp: Column | None = None) -> Column: - """Creates a condition column that checks if specified date or timestamp column is in the future. - Future is considered as grater than current timestamp plus `offset` seconds. +def is_not_in_future(col_name: str, offset: int = 0, curr_timestamp: Column | None = None) -> Column: + """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). :param col_name: column name :param offset: offset (in seconds) to add to the current timestamp at time of execution @@ -216,9 +218,10 @@ def not_in_future(col_name: str, offset: int = 0, curr_timestamp: Column | None ) -def not_in_near_future(col_name: str, offset: int = 0, curr_timestamp: Column | None = None) -> Column: - """Creates a condition column that checks if specified date or timestamp column is in the near future. - Near future is considered as grater than current timestamp but less than current timestamp plus `offset` seconds. +def is_not_in_near_future(col_name: str, offset: int = 0, curr_timestamp: Column | None = None) -> Column: + """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). :param col_name: column name :param offset: offset (in seconds) to add to the current timestamp at time of execution @@ -247,87 +250,59 @@ def not_in_near_future(col_name: str, offset: int = 0, curr_timestamp: Column | ) -def not_less_than(col_name: str, limit: int | datetime.date | datetime.datetime) -> Column: - """Creates a condition column that checks if a value is less than specified limit. +def is_not_less_than( + col_name: str, limit: int | datetime.date | datetime.datetime | str | Column | None = None +) -> Column: + """Checks whether the values in the input column are not less than the provided limit. :param col_name: column name - :param limit: limit to use in the condition + :param limit: limit to use in the condition as number, date, timestamp, column name or expression :return: new Column """ - limit_expr = F.lit(limit) + limit_expr = _get_column_expr_limit(limit) condition = F.col(col_name) < limit_expr return make_condition( condition, - F.concat_ws(" ", F.lit("Value"), F.col(col_name), F.lit("is less than limit:"), F.lit(limit).cast("string")), + F.concat_ws(" ", F.lit("Value"), F.col(col_name), F.lit("is less than limit:"), limit_expr.cast("string")), f"{col_name}_less_than_limit", ) -def not_greater_than(col_name: str, limit: int | datetime.date | datetime.datetime) -> Column: - """Creates a condition column that checks if a value is greater than specified limit. +def is_not_greater_than( + col_name: str, limit: int | datetime.date | datetime.datetime | str | Column | None = None +) -> Column: + """Checks whether the values in the input column are not greater than the provided limit. :param col_name: column name - :param limit: limit to use in the condition + :param limit: limit to use in the condition as number, date, timestamp, column name or expression :return: new Column """ - limit_expr = F.lit(limit) + limit_expr = _get_column_expr_limit(limit) condition = F.col(col_name) > limit_expr return make_condition( condition, - F.concat_ws(" ", F.lit("Value"), F.col(col_name), F.lit("is greater than limit:"), F.lit(limit).cast("string")), + F.concat_ws(" ", F.lit("Value"), F.col(col_name), F.lit("is greater than limit:"), limit_expr.cast("string")), f"{col_name}_greater_than_limit", ) -def _get_min_max_column_expr( - min_limit: int | datetime.date | datetime.datetime | str | None = None, - max_limit: int | datetime.date | datetime.datetime | str | None = None, - min_limit_col_expr: str | Column | None = None, - max_limit_col_expr: str | Column | None = None, -) -> tuple[Column, Column]: - """Helper function to create a condition for the is_(not)_in_range functions. - - :param min_limit: min limit value - :param max_limit: max limit value - :param min_limit_col_expr: min limit column name or expr - :param max_limit_col_expr: max limit column name or expr - :return: tuple containing min_limit_expr and max_limit_expr - :raises: ValueError when both min_limit/min_limit_col_expr or max_limit/max_limit_col_expr are null - """ - if (min_limit is None and min_limit_col_expr is None) or (max_limit is None and max_limit_col_expr is None): - raise ValueError('Either min_limit / min_limit_col_expr or max_limit / max_limit_col_expr is empty') - if min_limit_col_expr is None: - min_limit_expr = F.lit(min_limit) - else: - min_limit_expr = F.col(min_limit_col_expr) if isinstance(min_limit_col_expr, str) else min_limit_col_expr - if max_limit_col_expr is None: - max_limit_expr = F.lit(max_limit) - else: - max_limit_expr = F.col(max_limit_col_expr) if isinstance(max_limit_col_expr, str) else max_limit_col_expr - return (min_limit_expr, max_limit_expr) - - def is_in_range( col_name: str, - min_limit: int | datetime.date | datetime.datetime | str | None = None, - max_limit: int | datetime.date | datetime.datetime | str | None = None, - min_limit_col_expr: str | Column | None = None, - max_limit_col_expr: str | Column | None = None, + min_limit: int | datetime.date | datetime.datetime | str | Column | None = None, + max_limit: int | datetime.date | datetime.datetime | str | Column | None = None, ) -> Column: - """Creates a condition column that checks if a value is smaller than min limit or greater than max limit. + """Checks whether the values in the input column are in the provided limits (inclusive of both boundaries). :param col_name: column name - :param min_limit: min limit value - :param max_limit: max limit value - :param min_limit_col_expr: min limit column name or expr - :param max_limit_col_expr: max limit column name or expr + :param min_limit: min limit to use in the condition as number, date, timestamp, column name or expression + :param max_limit: max limit to use in the condition as number, date, timestamp, column name or expression :return: new Column """ - min_limit_expr, max_limit_expr = _get_min_max_column_expr( - min_limit, max_limit, min_limit_col_expr, max_limit_col_expr - ) + min_limit_expr = _get_column_expr_limit(min_limit) + max_limit_expr = _get_column_expr_limit(max_limit) + condition = (F.col(col_name) < min_limit_expr) | (F.col(col_name) > max_limit_expr) return make_condition( @@ -348,24 +323,20 @@ def is_in_range( def is_not_in_range( col_name: str, - min_limit: int | datetime.date | datetime.datetime | str | None = None, - max_limit: int | datetime.date | datetime.datetime | str | None = None, - min_limit_col_expr: str | Column | None = None, - max_limit_col_expr: str | Column | None = None, + min_limit: int | datetime.date | datetime.datetime | str | Column | None = None, + max_limit: int | datetime.date | datetime.datetime | str | Column | None = None, ) -> Column: - """Creates a condition column that checks if a value is within min and max limits. + """Checks whether the values in the input column are outside the provided limits (inclusive of both boundaries). :param col_name: column name - :param min_limit: min limit value - :param max_limit: max limit value - :param min_limit_col_expr: min limit column name or expr - :param max_limit_col_expr: max limit column name or expr + :param min_limit: min limit to use in the condition as number, date, timestamp, column name or expression + :param max_limit: min limit to use in the condition as number, date, timestamp, column name or expression :return: new Column """ - min_limit_expr, max_limit_expr = _get_min_max_column_expr( - min_limit, max_limit, min_limit_col_expr, max_limit_col_expr - ) - condition = (F.col(col_name) > min_limit_expr) & (F.col(col_name) < max_limit_expr) + min_limit_expr = _get_column_expr_limit(min_limit) + max_limit_expr = _get_column_expr_limit(max_limit) + + condition = (F.col(col_name) >= min_limit_expr) & (F.col(col_name) <= max_limit_expr) return make_condition( condition, @@ -384,7 +355,7 @@ def is_not_in_range( def regex_match(col_name: str, regex: str, negate: bool = False) -> Column: - """Creates a condition column to check if value not matches given regex. + """Checks whether the values in the input column matches a given regex. :param col_name: column name to check :param regex: regex to check @@ -402,8 +373,8 @@ def regex_match(col_name: str, regex: str, negate: bool = False) -> Column: def is_not_null_and_not_empty_array(col_name: str) -> Column: - """ - Creates a condition column to check if an array is null and or empty. + """Checks whether the values in the array input column are not null and not empty. + :param col_name: column name to check :return: Column object for condition """ @@ -413,8 +384,8 @@ def is_not_null_and_not_empty_array(col_name: str) -> Column: def is_valid_date(col_name: str, date_format: str | None = None) -> Column: - """ - Creates a condition column to check if a string is a valid date. + """Checks whether the values in the input column have valid date formats. + :param col_name: column name to check :param date_format: date format (e.g. 'yyyy-mm-dd') :return: Column object for condition @@ -433,8 +404,8 @@ def is_valid_date(col_name: str, date_format: str | None = None) -> Column: def is_valid_timestamp(col_name: str, timestamp_format: str | None = None) -> Column: - """ - Creates a condition column to check if a string is a valid timestamp. + """Checks whether the values in the input column have valid timestamp formats. + :param col_name: column name to check :param timestamp_format: timestamp format (e.g. 'yyyy-mm-dd HH:mm:ss') :return: Column object for condition @@ -452,3 +423,52 @@ def is_valid_timestamp(col_name: str, timestamp_format: str | None = None) -> Co F.concat_ws("", F.lit("Value '"), column, F.lit(condition_str)), f"{col_name}_is_not_valid_timestamp", ) + + +def is_unique(col_name: str, window_spec: str | Column | None = None) -> Column: + """Checks whether the values in the input column are unique + and reports an issue for each row that contains a duplicate value. + Null values are not considered duplicates, following the ANSI SQL standard. + It should be used carefully in the streaming context, + as uniqueness check will only be performed on individual micro-batches. + + :param col_name: column name to check + :param window_spec: window specification for the partition by clause. Default value for NULL in the time column + of the window spec must be provided using coalesce() to prevent rows exclusion! + e.g. "window(coalesce(b, '1970-01-01'), '2 hours')" + :return: Column object for condition + """ + column = F.col(col_name) + if window_spec is None: + partition_by_spec = Window.partitionBy(column) + else: + if isinstance(window_spec, str): + window_spec = F.expr(window_spec) + partition_by_spec = Window.partitionBy(window_spec) + + condition = F.when(column.isNotNull(), F.count(column).over(partition_by_spec) == 1) + return make_condition(~condition, f"Column {col_name} has duplicate values", f"{col_name}_is_not_unique") + + +def _cleanup_alias_name(col_name: str) -> str: + # avoid issues with structs + return col_name.replace(".", "_") + + +def _get_column_expr_limit( + limit: int | datetime.date | datetime.datetime | str | Column | None = None, +) -> Column: + """Helper function to generate a column expression limit based on the provided limit value. + + :param limit: limit to use in the condition (literal value or Column expression) + :return: column expression. + :raises ValueError: if limit is not provided. + """ + if limit is None: + raise ValueError("Limit is not provided.") + + if isinstance(limit, str): + return F.expr(limit) + if isinstance(limit, Column): + return limit + return F.lit(limit) diff --git a/src/databricks/labs/dqx/profiler/generator.py b/src/databricks/labs/dqx/profiler/generator.py index 2c51b518..2d1e6489 100644 --- a/src/databricks/labs/dqx/profiler/generator.py +++ b/src/databricks/labs/dqx/profiler/generator.py @@ -48,7 +48,7 @@ def dq_generate_is_in(col_name: str, level: str = "error", **params: dict): :return: A dictionary representing the data quality rule. """ return { - "check": {"function": "value_is_in_list", "arguments": {"col_name": col_name, "allowed": params["in"]}}, + "check": {"function": "is_in_list", "arguments": {"col_name": col_name, "allowed": params["in"]}}, "name": f"{col_name}_other_value", "criticality": level, } @@ -86,7 +86,7 @@ def dq_generate_min_max(col_name: str, level: str = "error", **params: dict): if max_limit is not None: return { "check": { - "function": "not_greater_than", + "function": "is_not_greater_than", "arguments": { "col_name": col_name, "val": val_maybe_to_str(max_limit, include_sql_quotes=False), @@ -99,7 +99,7 @@ def dq_generate_min_max(col_name: str, level: str = "error", **params: dict): if min_limit is not None: return { "check": { - "function": "not_less_than", + "function": "is_not_less_than", "arguments": { "col_name": col_name, "val": val_maybe_to_str(min_limit, include_sql_quotes=False), diff --git a/tests/conftest.py b/tests/conftest.py index f8dcda26..1c79fa9a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -22,7 +22,7 @@ def checks_yml_content(): trim_strings: true - criticality: warn check: - function: value_is_in_list + function: is_in_list arguments: col_name: col4 allowed: @@ -67,7 +67,7 @@ def checks_json_content(): { "criticality": "warn", "check": { - "function": "value_is_in_list", + "function": "is_in_list", "arguments": { "col_name": "col4", "allowed": [1, 2] @@ -130,7 +130,7 @@ def expected_checks(): }, { "criticality": "warn", - "check": {"function": "value_is_in_list", "arguments": {"col_name": "col4", "allowed": [1, 2]}}, + "check": {"function": "is_in_list", "arguments": {"col_name": "col4", "allowed": [1, 2]}}, }, { "criticality": "error", diff --git a/tests/integration/test_apply_checks.py b/tests/integration/test_apply_checks.py index 9b3a65d3..48c194e4 100644 --- a/tests/integration/test_apply_checks.py +++ b/tests/integration/test_apply_checks.py @@ -1,3 +1,6 @@ +from datetime import datetime + +import yaml import pyspark.sql.functions as F import pytest from pyspark.sql import Column @@ -213,14 +216,14 @@ def test_apply_checks_and_split_by_metadata(ws, spark): "check": {"function": "is_not_null_and_not_empty", "arguments": {"col_name": "c"}}, }, { - "name": "col_a_value_is_not_in_the_list", + "name": "col_a_is_not_in_the_list", "criticality": "warn", - "check": {"function": "value_is_in_list", "arguments": {"col_name": "a", "allowed": [1, 3, 4]}}, + "check": {"function": "is_in_list", "arguments": {"col_name": "a", "allowed": [1, 3, 4]}}, }, { - "name": "col_c_value_is_not_in_the_list", + "name": "col_c_is_not_in_the_list", "criticality": "warn", - "check": {"function": "value_is_in_list", "arguments": {"col_name": "c", "allowed": [1, 3, 4]}}, + "check": {"function": "is_in_list", "arguments": {"col_name": "c", "allowed": [1, 3, 4]}}, }, ] @@ -236,7 +239,7 @@ def test_apply_checks_and_split_by_metadata(ws, spark): None, 4, {"col_b_is_null_or_empty": "Column b is null or empty"}, - {"col_a_value_is_not_in_the_list": "Value 2 is not in the allowed list: [1, 3, 4]"}, + {"col_a_is_not_in_the_list": "Value 2 is not in the allowed list: [1, 3, 4]"}, ], [ None, @@ -280,7 +283,7 @@ def test_apply_checks_and_split_by_metadata_with_autogenerated_col_names(ws, spa }, { "criticality": "warn", - "check": {"function": "value_is_in_list", "arguments": {"col_names": ["a", "c"], "allowed": [1, 3, 4]}}, + "check": {"function": "is_in_list", "arguments": {"col_names": ["a", "c"], "allowed": [1, 3, 4]}}, }, ] @@ -296,7 +299,7 @@ def test_apply_checks_and_split_by_metadata_with_autogenerated_col_names(ws, spa None, 4, {"col_b_is_null_or_empty": "Column b is null or empty"}, - {"col_a_value_is_not_in_the_list": "Value 2 is not in the allowed list: [1, 3, 4]"}, + {"col_a_is_not_in_the_list": "Value 2 is not in the allowed list: [1, 3, 4]"}, ], [ None, @@ -340,7 +343,7 @@ def test_apply_checks_by_metadata(ws, spark): }, { "criticality": "warn", - "check": {"function": "value_is_in_list", "arguments": {"col_names": ["a", "c"], "allowed": [1, 3, 4]}}, + "check": {"function": "is_in_list", "arguments": {"col_names": ["a", "c"], "allowed": [1, 3, 4]}}, }, ] @@ -354,7 +357,7 @@ def test_apply_checks_by_metadata(ws, spark): None, 4, {"col_b_is_null_or_empty": "Column b is null or empty"}, - {"col_a_value_is_not_in_the_list": "Value 2 is not in the allowed list: [1, 3, 4]"}, + {"col_a_is_not_in_the_list": "Value 2 is not in the allowed list: [1, 3, 4]"}, ], [ None, @@ -751,3 +754,343 @@ def test_apply_checks_with_sql_expression(ws, spark): expected_schema, ) assert_df_equality(checked, expected, ignore_nullable=True) + + +def test_apply_checks_with_is_unique(ws, spark, set_utc_timezone): + schema = "col1: int, col2: timestamp" + test_df = spark.createDataFrame([[1, datetime(2025, 1, 1)], [1, datetime(2025, 1, 2)], [None, None]], schema) + + checks = [ + { + "criticality": "error", + "check": {"function": "is_unique", "arguments": {"col_name": "col1"}}, + }, + { + "criticality": "error", + "name": "col_col1_is_not_unique2", + "check": { + "function": "is_unique", + "arguments": {"col_name": "col1", "window_spec": "window(coalesce(col2, '1970-01-01'), '30 days')"}, + }, + }, + ] + + dq_engine = DQEngine(ws) + checked = dq_engine.apply_checks_by_metadata(test_df, checks) + + expected_schema = schema + REPORTING_COLUMNS + expected = spark.createDataFrame( + [ + [None, None, None, None], + [ + 1, + datetime(2025, 1, 1), + { + "col_col1_is_not_unique": "Column col1 has duplicate values", + "col_col1_is_not_unique2": "Column col1 has duplicate values", + }, + None, + ], + [ + 1, + datetime(2025, 1, 2), + { + "col_col1_is_not_unique": "Column col1 has duplicate values", + "col_col1_is_not_unique2": "Column col1 has duplicate values", + }, + None, + ], + ], + expected_schema, + ) + assert_df_equality(checked, expected, ignore_nullable=True) + + +def test_apply_checks_all_checks_as_yaml(ws, spark): + checks = yaml.safe_load( + """ + # is_not_null check + - criticality: error + check: + function: is_not_null + arguments: + col_name: col1 + + # is_not_empty check + - criticality: error + check: + function: is_not_empty + arguments: + col_name: col1 + + # is_not_null_and_not_empty check + - criticality: error + check: + function: is_not_null_and_not_empty + arguments: + col_name: col1 + trim_strings: true + + # is_in_list check + - criticality: error + check: + function: is_in_list + arguments: + col_name: col2 + allowed: + - 1 + - 2 + - 3 + + # is_not_null_and_is_in_list check + - criticality: error + check: + function: is_not_null_and_is_in_list + arguments: + col_name: col2 + allowed: + - 1 + - 2 + - 3 + + # is_not_null_and_not_empty_array check + - criticality: error + check: + function: is_not_null_and_not_empty_array + arguments: + col_name: col4 + + # is_in_range check + - criticality: error + check: + function: is_in_range + arguments: + col_name: col2 + min_limit: 1 + max_limit: 10 + - criticality: error + check: + function: is_in_range + arguments: + col_name: col5 + min_limit: 2025-01-01 + max_limit: 2025-02-24 + - criticality: error + check: + function: is_in_range + arguments: + col_name: col6 + min_limit: 2025-01-01 00:00:00 + max_limit: 2025-02-24 01:00:00 + - criticality: error + check: + function: is_in_range + arguments: + col_name: col3 + min_limit: col2 + max_limit: col2 * 2 + + # is_not_in_range check + - criticality: error + check: + function: is_not_in_range + arguments: + col_name: col2 + min_limit: 11 + max_limit: 20 + - criticality: error + check: + function: is_not_in_range + arguments: + col_name: col5 + min_limit: 2025-02-25 + max_limit: 2025-02-26 + - criticality: error + check: + function: is_not_in_range + arguments: + col_name: col6 + min_limit: 2025-02-25 00:00:00 + max_limit: 2025-02-26 01:00:00 + - criticality: error + check: + function: is_not_in_range + arguments: + col_name: col3 + min_limit: col2 + 10 + max_limit: col2 * 10 + + # is_not_less_than check + - criticality: error + check: + function: is_not_less_than + arguments: + col_name: col2 + limit: 0 + - criticality: error + check: + function: is_not_less_than + arguments: + col_name: col5 + limit: 2025-01-01 + - criticality: error + check: + function: is_not_less_than + arguments: + col_name: col6 + limit: 2025-01-01 01:00:00 + - criticality: error + check: + function: is_not_less_than + arguments: + col_name: col3 + limit: col2 - 10 + + # is_not_greater_than check + - criticality: error + check: + function: is_not_greater_than + arguments: + col_name: col2 + limit: 10 + - criticality: error + check: + function: is_not_greater_than + arguments: + col_name: col5 + limit: 2025-03-01 + - criticality: error + check: + function: is_not_greater_than + arguments: + col_name: col6 + limit: 2025-03-24 01:00:00 + - criticality: error + check: + function: is_not_greater_than + arguments: + col_name: col3 + limit: col2 + 10 + + # is_valid_date check + - criticality: error + check: + function: is_valid_date + arguments: + col_name: col5 + - criticality: error + name: col5_is_not_valid_date2 + check: + function: is_valid_date + arguments: + col_name: col5 + date_format: yyyy-MM-dd + + # is_valid_timestamp check + - criticality: error + check: + function: is_valid_timestamp + arguments: + col_name: col6 + timestamp_format: yyyy-MM-dd HH:mm:ss + - criticality: error + name: col6_is_not_valid_timestamp2 + check: + function: is_valid_timestamp + arguments: + col_name: col6 + + # is_not_in_future check + - criticality: error + check: + function: is_not_in_future + arguments: + col_name: col6 + offset: 86400 + + # is_not_in_near_future check + - criticality: error + check: + function: is_not_in_near_future + arguments: + col_name: col6 + offset: 36400 + + # is_older_than_n_days check + - criticality: error + check: + function: is_older_than_n_days + arguments: + col_name: col5 + days: 10000 + + # is_older_than_col2_for_n_days check + - criticality: error + check: + function: is_older_than_col2_for_n_days + arguments: + col_name1: col5 + col_name2: col6 + days: 2 + + # is_unique check + - criticality: error + check: + function: is_unique + arguments: + col_name: col1 + - criticality: error + name: col1_is_not_unique2 + check: + function: is_unique + arguments: + col_name: col1 + window_spec: window(coalesce(col6, '1970-01-01'), '10 minutes') + + # regex_match check + - criticality: error + check: + function: regex_match + arguments: + col_name: col2 + regex: '[0-9]+' + negate: false + + # sql_expression check + - criticality: error + check: + function: sql_expression + arguments: + expression: col3 > col2 and col3 < 10 + msg: col3 is greater than col2 and col3 less than 10 + name: custom_output_name + negate: false + """ + ) + + dq_engine = DQEngine(ws) + status = dq_engine.validate_checks(checks) + assert not status.has_errors + + schema = "col1: string, col2: int, col3: int, col4 array