Uniquess Check

[rishujuneja1298]

I have been trying to create a custom check functions to perform the unique value check for a column. But I am executing errors using the GroupBy clause. Is there any limitations there?

[mwojtyczka]

@rishujuneja1298 No really as long as you are able to return make_condition. To check uniqueness you could use a window function. We should add this as a standard check:

def is_unique(col_name: str) -> Column:
    """Creates a condition column to check if values in a column are unique.

    :param col_name: column name to check
    :return: Column object for condition
    """
    window_spec = Window.partitionBy(F.col(col_name))
    condition = F.count(F.col(col_name)).over(window_spec) == 1

    return make_condition(~condition, f"Column {col_name} has duplicate values", f"{col_name}_is_not_unique")

This will report issue for every row that contains a duplicate for the column.

[rishujuneja1298]

Thanks for the feedback, it is working. But there is one thing if I want to call the custom defined checks we need have a argument globals() which will take this custom defined check.

In my case. I have defined all my checks in a .yml file which contains some other predefined checks also.

• check:
  arguments:
  col_names:
  - message
  - project
  - productId
  function: is_not_null
  criticality: error

• check:
  arguments:
  col_names:
  - project
  regex: "^(?=.[a-zA-Z])[a-zA-Z0-9 _-]$"
  negate: false
  function: regex_match
  criticality: error

• check:
  arguments:
  col_names:
  - productId
  function: is_unique
  criticality: error

And I am calling the below function to run the dqx. If I am including globals() it fails with errors like "Error applying checks to dataframe: function 'is_not_null' is not defined: {'check':" and if I excluded globals it failed with error " ERROR [root] Error applying checks to dataframe: function 'is_unique' is not defined:". How can we handle this case?

def apply_checks_by_metadata(input_df, check_path: str):
if input_df is None or input_df.rdd.isEmpty():
raise Exception("Input dataframe is invalid or empty.")

# Load checks from the given path
checks = load_checks(check_path)

try:
    # Apply checks and return a single dataframe containing both valid and quarantined data
    valid_and_quarantined_df = dq_engine.apply_checks_by_metadata(input_df, checks,globals())
    logging.info("Successfully applied checks to the dataframe.")
    return valid_and_quarantined_df
except Exception as e:
    logging.error(f"Error applying checks to dataframe: {e}")
    raise Exception(f"Error applying checks to dataframe: {e}")

[mwojtyczka]

Great. The uniquness check should be released this week: #200

The error is expected. Before calling apply_checks_by_metadata you must import the functions that you use (either individual or all):
from databricks.labs.dqx.col_functions import *

I agree it's a bit cumbersome so I am changing this so that predefined functions are included by default without a need to make import: #203