Data Quality (DQ) Processor using Unity Catalog

Calibo Accelerate supports end-to-end data quality management for data stored in Unity Catalog by integrating data ingestion, data profiler, data analyzer, data validator, and issue resolver into a structured pipeline.

To simplify data quality implementation and improve operational efficiency, data ingestion and data profiling are combined into a single Data Integration stage, while data analyzer, data validator and issue resolver are handled in a separate stage of data quality called DQ Processor.

In the data integration stage, data is ingested from source systems such as Microsoft SQL Server using Databricks for data integration and loaded into Databricks Unity Catalog. As part of the same stage, data profiling is performed to analyze structural and statistical characteristics of the ingested data, helping establish baseline quality metrics and metadata early in the pipeline.

The DQ Processor stage focuses on applying validation rules, identifying data quality issues, and managing issue resolver workflows based on the profiling results and defined constraints.

This topic describes how to create a data quality processor pipeline that reads data from Databricks Unity Catalog, applies data quality rules to validate and resolve identified issues, and writes the processed results back to Databricks Unity Catalog for downstream use.

To create a DQ Processor job, you must complete the following high-level steps:

  1. Create a DQ processor job giving it an appropriate name.

  2. Select the source table and data processing type.

  3. Enable Rule Configuration.

    1. Generate rules by running rule suggestions job and notify users via email.

    2. View and add rules.

    3. Test rules on actual data.

    4. View rule results.

    5. Add analyzer rules as required.

    6. Add reference and schema based on your use case.

    7. Click Run Validations.

    8. Check the validation results.

  4. Enable Issue Resolver

    1. Select Data Processing Mode:

      • Passed Validation Data

      • Full Data

    2. Select the constraints to run on the data.

  5. Select the target schema and tables to store successful records, Validator records and rejected Issue Resolver records.

The data quality processor pipeline has the following nodes:

Databricks Unity Catalog (data lake) > Databricks Unity Catalog (data quality node) - DQ Processor

Prerequisites

To create or run a data quality processor job using Unity Catalog, you must complete the following prerequisites:

  • Get access to a Unity Catalog data lake configuration listed under Configuration > Cloud Platform Tools & Technologies > Databases and Data Warehouses.

  • Ensure that the Statistics Common Table is configured in the Unity Catalog instance to store statistical data and metadata information.

  • Ensure that the Databricks cluster is configured with a Databricks Machine Learning Runtime before enabling anomaly detection.

To create a data quality processor job

  1. Sign in to the  Calibo Accelerate platform and navigate to Products.

  2. Select a product and feature. Click the Develop stage of the feature and navigate to Data Pipeline Studio.

  3. Add the Data Lake stage > Databricks Unity Catalog node, then configure the data lake node.

  4. Add the Data Quality stage > Databricks Unity Catalog > DQ Processor node.

  5. Connect the data lake and data quality nodes to each other.

  6. Click the data quality processor node and complete the following steps to create a data quality processor job:

    Provide job details for the data quality processor job:

    • Template - Based on the source and destination that you choose in the data pipeline, the template is automatically selected.

    • Job Name - Provide an appropriate name for the data quality processor job.

    • Node rerun Attempts - Specify the number of times the pipeline rerun is attempted on this node of the pipeline, in case of failure. The default setting is done at the pipeline level. You can change the rerun attempts by selecting 1,2, or 3.

    • Fault Tolerance - Define the behavior of the node upon failure, where the descendent nodes can either stop and skip execution or can continue their normal operation. The available options are:

      • Default - If a node fails, the subsequent nodes go into pending state.

      • Proceed on Failure - If a node fails, the subsequent nodes are executed.

      • Skip on Failure - If a node fails, the subsequent nodes are skipped from execution.

        For more information, see Fault Tolerance of Data Pipelines

    • Source - This is automatically selected depending on the data lake node configured and added in the pipeline.

    • Datastore - The configured datastore that you added in the data pipeline is displayed.

    • Catalog Name - The catalog which is associated with the configured datastore is displayed.

    • Schema Name - The schema associated with the catalog is displayed. The schema is selected based on the catalog, but you can select a different schema.

    • Source Table - Select a table from the selected datastore.

    • Data Processing Type - Select the type of processing that must be done for the data. Choose from the following options:

      • Delta - In this type of processing, incremental data is processed. For the first job run, the complete data is considered. For subsequent job runs, only delta data is considered.

        • Based On - The delta processing is done based on the following options:

          Option Description
          Table Versions (Change Data Feed)

          • Unity Catalog stores data in delta format, this feature is called Change Feed Data. Unity Catalog creates versions of tables when there is a change in data. Select this option to automatically process incremental changes from Delta tables.

          • Unique Identifier Columns - Select a column (or set of columns) in the Databricks Delta Table that uniquely identifies each record and can be used as a reference for retrieving the latest version of the data.
          Table Columns

          • Select this option to select a column manually for processing of data from Delta tables.

          • Delta Identification Columns - Select columns used for delta identification, such as date or incremental ID. These columns help determine which records have been modified or added since the last data load.

      • Full - In this type of processing, the complete dataset is processed in each job run.

      Click Next.

    This is an optional step if you want to add rules and run validation. To configure rules, do the following:

    1. In the Rule Configuration step, turn on the Enable Rule Configuration toggle. Enabling Rule Configuration allows you to define, test, and apply data quality rules to the selected dataset.

    2. In the Get started with Generate Data Quality Rules section, click Generate Data Quality Rules and Notify. The Run Rule Suggestions popup opens.

      1. You can do either of the following:

        • Run the Generate Data Quality Rules job on the full dataset.

        • Enable Apply Filter to choose fields, operators, and values and preview the filter details. You can construct complex filter conditions using AND/OR combinations, validate the filter query, and apply SQL WHERE‑clause filters to limit the dataset used for generating and validating rules.

          Run Rule Suggestions_Apply Filter

        • Enter a number in Limit the records (Optional) to reduce the data size.

      2. Select Send me an email when rule generation is complete, then click Generate Rules & Notify.

        The Get started with Generate Data Quality Rules section displays the progress bar when the Generate Data Quality Rules job is initiated, a confirmation message is displayed and email notifications are sent once the rule generation is completed successfully.

        Note:

        • You can click the Refresh icon to check the status or click Cancel to terminate the initiated job.

        • To regenerate data quality rules and notify users of the results, you can click the Regenerate Rule Suggestion and Notify option.

        • If rules were previously generated, this section displays Last Run Status and Completed On date and time.

        The dataset is analyzed and recommended data quality rule suggestions are generated based on data patterns and profiling insights.

      1. Click View & Add Rules. The Add Rule Suggestions popup opens.

      1. From the Select Completed Run Suggestions to View dropdown, select the completed job from the available list as needed.

      1. Review the list of generated rules and select the required rules based on your data quality requirements.

        Tip: You can choose only the rules relevant to your use case and ignore others.

      2. Select a severity level (Low, Medium, High) from the dropdown for each chosen rule. Severity indicates the rule’s importance and is used during data quality evaluation and pipeline processing.

      1. Click Add to add the selected rules to the configuration.

        The selected rules appear in the Code Editor section.

      Note:

      For validating or troubleshooting rule generation outcomes, you can do the following:

      • From the ellipsis, click View Run Details to review following execution details in the Rule Generation Run Details popup:

        • Total Runs

        • Ongoing Runs

        • Cancelled Runs

        • Failed Runs

        • Successful Runs

        • Rule Generation Run History including Filter Criteria, Status, Start Time, End Time, Run Duration, and Rule Output, where you can copy the generated rules.

      • Click Open Databricks Dashboard to open the Databricks dashboard associated with rule analysis and execution.

    3. To add or modify rules manually using code editor, do the following:

      1. In the Code Editor section, edit the rules using supported rule syntax and check for Errors/Warnings, then click Test Rules on Source Data to validate rules. The Do you want to run rules on the actual dataset? popup opens.

        Alert: Resolve errors before proceeding to ensure successful validation.

      2. In the popup, you can do either of the following:

        • Run rules job on the full dataset.

        • Enter a percentage in the Limit the records to process by specifying a percentage textbox to reduce the data size.

      3. Select Send me an email when the rule execution is complete, then click Yes.

      The Code Editor section displays the progress bar when the Data Quality Run validation job is initiated, and a confirmation message is displayed once the validation runs successfully.

      Note:

      You can click the Refresh icon to check the status or click the Stop icon to terminate the initiated job.

      1. Click View Rule Results to review the following validation details in the Results popup:

        • Percentage of records processed

        • Analyzer tab including details of analysed rules

        • Rules (Validator) tab including Rules, Result (Pass/Fail), and Result Details.

        Note:

        For validating or troubleshooting rule generation outcomes, you can do the following:

        • From the ellipsis, click Run Validation to run rules on the actual dataset. This option is available only if more rules are added after generating first set of rules.

        • Click Open Databricks Dashboard to open the Databricks dashboard associated with rule validation.

    4. To define and apply validation rules that ensure data accuracy based on specific constraints, do the following:

      1. In the Validator Rules tab, use Search Constraints to locate the required validation rules, or browse through the following categories:

        • Aggregated - Constraints applied on grouped or summary data.

        • Non-aggregated - Row-level constraints applied to individual records.

      2. Select the required validator rules and severity (Low, Medium, High) as applicable, apply filters if needed, then click Add. The selected validator rules appear in the Code Editor section.

        3. Note:

        When applying a filter, you can select the field, choose an operator, enter a value, and preview the filter details. Additionally, you can restrict the dataset used for rule validation by applying SQL WHERE‑clause filters directly. This allows you to validate only a specific subset of the data based on custom filtering conditions.

    5. To configure analyzer rules that help assess data health including data characteristics and statistics, do the following:

      1. Click the Analyzer Rules tab, use Search Constraints to locate specific analyzer rules.

      2. Select the required analyzer rules. A popup opens.

      3. Turn on Anomaly or Enable Auto-Anomaly option to enable anomaly detection for the selected analyzer rules.

        1. You can enable anomaly detection for each rule individually or for all rules simultaneously.

        2. Anomaly settings are not considered when you click Test Rules on Source Data.

        3. Anomaly detection is evaluated only during DQ Processor pipeline execution.

        4. The anomaly detection model requires sufficient historical execution data before anomaly predictions become available.

      4. Enter an anomaly Weightage value to indicate the relative importance of the analyzer metric during anomaly evaluation. Anomaly weightage determines the relative importance assigned to an analyzer metric during anomaly evaluation.

        • Higher values increase the contribution of the metric during anomaly scoring.

        • Lower values reduce the contribution of the metric.

        • If you do not specify a weightage, the system uses the default weightage during anomaly calculations.

      5. Click Add. The selected analyzer rules are added to the Code Editor section with anomaly detection settings.

        Example: Completeness "trade_id" with Anomaly = “true” with Anomaly_weightage = “0.5”

        In the Code Editor, analyzer rules configured for anomaly detection display:

        • Rule name

        • Column name (if applicable)

        • Anomaly enabled status

        • Anomaly weightage value (when specified)

    6. To add schema and reference tables that help validate data against a master dataset with predefined list of allowed values, do the following:

      1. Click on the Schema & References tab, use Search Schema to locate available schemas or reference tables.

      2. Click Add Reference Table to include a reference table for validating data against external or master datasets. The selected schema and reference tables appear in the Code Editor section. For more information, refer to Understanding Schema and Reference Tables.

      3. In the reference table side‑drawer, click Updates Available to view column additions or deletions and update the reference table accordingly. When modifying reference table entries, the system performs usage validation to ensure that these changes do not break existing configurations or dependent data quality rules.

        • If a referenced value is used in rules or mappings, an alert message displays before allowing the update or deletion.

        • Only safe, non‑breaking changes are permitted, ensuring the integrity of all dependent DQ rules and validations.

        Updates Available_Schema & References

    7. Enable Abort pipeline run if the validator results fail to stop the pipeline execution when validation rules fail. This option ensures only validated data is processed in the pipeline.

    8. Click Next.

    To apply automated data handling constraints and resolve issues, do the following:

    1. In the Issue Resolver step, turn on the Enable Issue Resolver toggle. This allows you to apply automated rules to resolve issues during data processing.

    2. From Select Data Processing Mode, choose one of the following:

      • Passed Validation Data - Applies issue resolver only to records that passed validation.

        Note:

        This option is available only if the Enable Rule Configuration option is enabled in the Rule Configuration step.

      • Full Data - Applies issue resolver to all records.

      Tip: Choose the mode based on whether you want strict or inclusive processing.

    3. To configure data issue resolver constraints that determine how specific data quality issues are automatically handled during processing, do the following:

      1. In the Data Issue Resolver Constraints section, configure column-level issue handling using the following options:

        • Column Name - Displays the column being configured to which issue resolution rules are applied.

        • Handle Duplicate Data - Define how duplicate records are detected and resolved for the column.

        • Unique Key Columns - Select columns that uniquely identify records for duplicate detection.

        • Duplicate Data Order - Specify the ordering logic used for resolving duplicate records.

        • Handle Missing Data - Define how null or empty values should be handled for the column.

        • Handle Outliers - Configure rules to identify and resolve outlier values in the column.

        • Handle String Operation - Apply string-level transformations or cleanup operations on column values.

        • Handle Case Sensitivity - Define how case sensitivity should be handled for string comparisons or values.

        • Replace Selective Data - Configure rules to replace specific values in the column.

        • Handle Data Against Master Table - Validate or resolve column values using a master reference table.

      2. Drag and drop to change the execution order of constraint columns.

    4. Review all configured issue resolution rules. Once configured, validation rules assess data quality, issue resolver rules automatically handle detected issues, and clean, validated data is written to the configured target tables.

    5. Click Next.

    This step defines where processed data and data quality outcomes are written after validation and issue resolution.

    1. In the Target step, review the following:

      • Target - This field is automatically populated, depending on the target node configured and added in the pipeline.

      • Datastore - This field is automatically populated, depending on the target node configured and added in the pipeline.

      • Catalog Name - This field is automatically populated, depending on the catalog selected for the selected Unity Catalog.

      • Schema Name - Select the schema from the dropdown list within the selected catalog where output tables will be created.

      These fields determine the logical storage location for all target tables.

    2. To define where records that pass validation and issue resolution are written:

      1. Under Successful Records do the following:

        • Select Schema Name for successful output records.

        • Select or enter the Target Table name where successful records will be stored.

        Successful records typically represent clean, compliant data ready for downstream consumption.

    3. To capture records that fail validation rules, enable the Failed Validator Records toggle and do the following:

      1. Select Schema Name for failed validator records.

      2. Select or enter the Target Table name where failed validator records will be stored.

      Note:

      Select Add Validation Columns to create an additional column in the output table to log validation errors for failed or rejected records.

    4. To store records that are rejected during issue resolution, enable the Rejected Issue Resolver Records toggle and do the following:

      1. Select Schema Name for rejected records.

      2. Select or enter the Target Table name where rejected issue resolver records will be stored.

      Note:

      You can select the Add Validation Columns to create an additional column in the output table to log validation errors for failed or rejected records.

    5. Statistics Repository Table - Displays the statistics repository table configured for the selected Databricks instance. This table is used for monitoring, auditing, and reporting data quality trends.

      After completing the Target step, clean records are written to designated target tables, failed and rejected records are optionally stored for analysis, and data quality statistics are captured for monitoring and governance.

    6. Click Next.

    In this step you provide schema mapping details for all mapped tables.

    1. Do the following:

      • Mapped Data - Select the mapping of the source file with the target table configured in the previous stage.

      • Infer Source Schema - This option lets you identify the schema of columns with data types automatically.

        Note:

        If this setting is turned off, you can manually edit the data types of source columns.

      • Auto - Evolve Target Schema - This option lets the integration job automatically update the target table by adding any new columns detected in the source schema.

      • Ignore New Columns - Enable this option to skip newly added columns and run the pipeline successfully.

      • Filter columns from selected table - Deselect columns you want to exclude from the mapping, configure constraints for required fields, or secure PII for columns containing sensitive information.

        • Click Column Updates Available to review and update any schema changes such as column additions or deletions detected in the source table. When schema updates occur, the system warns you about changes that may impact existing data quality rules in the Rule Configuration and Schema Mapping steps.

          Column Updates Available_Schema Mapping

          Column Updates Available_Popup

        • Deselect any columns that are not required from the auto‑populated list and assign custom names to columns where needed.

        • Apply constraints by selecting one of the available options:

          • Set Not Null

          • Check - for this constraint, you must provide a valid SQL condition and specify the value to be checked for the selected column.

      • Continue job run even if constraints are not met - Enable this option to ensure that the job run is continued even if a constraint is not met.

      • Add Custom Columns - Enable this option to add additional columns apart from the existing columns of the table. To add custom columns, do the following:

        1. Column Name - Provide a column name for the custom column that you want to add.

        2. Type and Value - Select the parameter type for the new column. Choose from the following options:

          • Static Parameter - Provide a static value that is added for this column.

          • System Parameter - Select a system-generated parameter from the dropdown list that must be added to the custom column.

          • Generated- Provide the SQL code to combine two or more columns to generate the value of the new column.

        3. Click Add Custom Column after adding the details for each column.

        4. Repeat steps 1-3 for the number of columns that you want to add. After adding the required custom columns, click Add Schema Mapping.

    2. To review the column mapping details, click the ellipsis (...) and click View Details.

      Note:

      You can use the Edit or Delete options to modify or remove the details.

    3. Click Next.

    In this step you select the operation type that you want to perform on the source table and the partitioning that you want to create on the target table.

    • Mapped Data - Select the mapping that you performed in the previous step. The source table and target table details are provided.

    • Operation Type - Select the operation type to perform on the source data. Choose one of the following options:

      • Append - Adds new data at the end of a file without erasing the existing content.

      • Merge - Adds data to the target table for the first job run. For each subsequent job run, the data from the target table is merged with the change in source data.

      • Overwrite - Replaces the entire content of a file with new data.

    • Enable Partitioning - Enable this option if you want to use partitioning for the target data. Select from the following options:

      • Data Partition - Select column name from the dropdown list. Select a column from the dropdown list. Click Add.

      • Date Based Partition - Select the type from the options - yearly, monthly, or daily. Provide the prefix that you would like to add to the partition. Adding a prefix is optional.

      Note:

      You cannot enable partitioning for an existing target table which does not have partitioning enabled.

    • Click Add. The Data mapping for mapped tables displays the mapping details. Click the ellipsis (...) to edit or delete the mapping.

    Click Next.

    You can select an all-purpose cluster or a job cluster to run the configured job. In case your Databricks cluster is not created through the Calibo Accelerate platform and you want to update custom environment variables, refer to the following:

    Updating Custom Variables for a Databricks Cluster

    Note:

    To use anomaly detection for analyzer rules, you must run the DQ processor job on an ML-enabled Databricks cluster. The Cluster Configuration step shows a notification when an anomaly detection is enabled.

    Cluster - Select the all-purpose cluster that you want to use for the data integration job, from the dropdown list.

    Note:

    If you do not see a cluster configuration in the dropdown list, it is possible that the configured Databricks cluster has been deleted.

    In this case, you must create a new Databricks cluster configuration in the Data Integration section of Cloud Platform Tools and Technologies. Delete the data integration node from the data pipeline, add a new node with the newly created configuration, and configure the job again. Now the user can select the newly configured Databricks cluster.

    Cluster Details Description
    Choose Cluster Provide a name for the job cluster that you want to create.
    Job Configuration Name Provide a name for the job cluster configuration.
    Databricks Runtime Version Select the appropriate Databricks version.
    Worker Type Select the worker type for the job cluster.
    Workers

    Enter the number of workers to be used for running the job in the job cluster.

    You can either have a fixed number of workers or you can choose autoscaling.
    Enable Autoscaling Autoscaling helps in scaling up or down the number of workers within the range specified by you. This helps in reallocating workers to a job during its compute-intensive phase. Once the compute requirement reduces the excess number of workers are removed. This helps control your resource costs.
    Cloud Infrastructure Details
    First on Demand

    Provide the number of cluster nodes that are marked as first_on_demand.

    The first_on_demand nodes of the cluster are placed on on-demand instances.

    Availability

    Choose the type of EC2 instances to launch your Apache Spark clusters, from the following options:

    • Spot

    • On-demand

    • Spot with fallback
    Zone

    Identifier of the availability zone or data center in which the cluster resides.

    The provided availability zone must be in the same region as the Databricks deployment.
    Instance Profile ARN Provide an instance profile ARN that can access the target Amazon S3 bucket.
    EBS Volume Type The type of EBS volume that is launched with this cluster.
    EBS Volume Count The number of volumes launched for each instance of the cluster.
    EBS Volume Size The size of the EBS volume to be used for the cluster.
    Additional Details
    Spark Config To fine tune Spark jobs, provide custom Spark configuration properties in key value pairs.
    Environment Variables Configure custom environment variables that you can use in init scripts.
    Logging Path (DBFS Only) Provide the logging path to deliver the logs for the Spark jobs.
    Init Scripts Provide the init or initialization scripts that run during the start up of each cluster.

    You can select an all-purpose cluster or a job cluster to run the configured job. In case your Databricks cluster is not created through the Calibo Accelerate platform and you want to update custom environment variables, refer to the following:

    Updating Custom Variables for a Databricks Cluster

    Cluster - Select the all-purpose cluster that you want to use for the data integration job, from the dropdown list.

    Cluster DetailsDescription
    Choose ClusterProvide a name for the job cluster that you want to create.
    Job Configuration NameProvide a name for the job cluster configuration.
    Databricks Runtime VersionSelect the appropriate Databricks version.
    Worker TypeSelect the worker type for the job cluster.
    Workers

    Enter the number of workers to be used for running the job in the job cluster.

    You can either have a fixed number of workers or you can choose autoscaling.

    Enable AutoscalingAutoscaling helps in scaling up or down the number of workers within the range specified by you. This helps in reallocating workers to a job during its compute-intensive phase. Once the compute requirement reduces the excess number of workers are removed. This helps control your resource costs.
    Cloud Infrastructure Details
    First on Demand

    Provide the number of cluster nodes that are marked as first_on_demand.

    The first_on_demand nodes of the cluster are placed on on-demand instances.

    Availability

    Choose the type of EC2 instances to launch your Apache Spark clusters, from the following options:

    • Spot

    • On-demand

    • Spot with fallback

    Zone

    Identifier of the availability zone or data center in which the cluster resides.

    The provided availability zone must be in the same region as the Databricks deployment.

    Instance Profile ARNProvide an instance profile ARN that can access the target Amazon S3 bucket.
    EBS Volume TypeThe type of EBS volume that is launched with this cluster.
    EBS Volume CountThe number of volumes launched for each instance of the cluster.
    EBS Volume SizeThe size of the EBS volume to be used for the cluster.
    Additional Details
    Spark ConfigTo fine tune Spark jobs, provide custom Spark configuration properties in key value pairs.
    Environment VariablesConfigure custom environment variables that you can use in init scripts.
    Logging Path (DBFS Only)Provide the logging path to deliver the logs for the Spark jobs.
    Init ScriptsProvide the init or initialization scripts that run during the start up of each cluster.

    Click Next.

    You can configure the SQS and SNS services to send notifications related to the node in this job. This provides information about various events related to the node without connecting to the Calibo Accelerate platform.

    SQS and SNS
    Configurations - Select an SQS or SNS configuration that is integrated with the Calibo Accelerate platform.

    Events - Enable the events for which you want to enable notifications:

    • Select All

    • Node Execution Failed

    • Node Execution Succeeded

    • Node Execution Running

    • Node Execution Rejected
    Event Details - Select the details of the events from the dropdown list, that you want to include in the notifications.
    Additional Parameters - Provide any additional parameters that are to be added in the SQS and SNS notifications. A sample JSON is provided, you can use this to write logic for processing the events.

 

Related Topics Link IconRecommended Topics What's next? Databricks Templatized Data Integration Jobs