dqu 0.2.5
Build intelligent, efficient, and trustable Data & ML Engineering Pipelines with this enterprise-ready Data Quality framework.
Requires Python: >=3.7

# DQU

**Build intelligent, efficient, and trustable Data & ML Engineering Pipelines with this enterprise-ready Data Quality framework**

<p align="center">
  <img width="200" alt="Dqu-Clean" src="https://github.com/user-attachments/assets/e5394d0b-dcb3-4f71-a280-d856ee550a97" />
</p>

## What is DQU?

DQU (Data QUality) is a unified framework that empowers data and machine learning developers to perform **automated data quality assessments** on **structured** or **semi-structured** data across modern distributed computing engines.

It supports:

- **Pandas** workflows
- **Apache Spark** for distributed batch/stream processing
- **Apache Flink** for stream processing
- **Ray Datasets** for distributed Python-native processing
- **Spark, Pandas, Ray** through a YAML config
- Supports YAML Driven Configuration
- Supports (Basic and Advanced Evals), Advanced emitts Error Records Dataframe along with eval results
- Supports Logging integrations with GCP Cloud Logging, Azure Log Analytics, Console Logs

DQU empowers working on Any cloud GCP, Azure (Microsoft Fabric, Azure Databricks, Azure Synapse) or AWS or local Jupyter Python environment, on any of these 4 engines (Pandas, Spark, Flink, Ray), DQU seamlessly integrates into your data & machine learning pipelines.

---

## GitHub

https://github.com/keshavksingh/dqu

---

## 📚 Table of Contents

- [What is DQU?](#what-is-dqu)
- [Key Features](#-key-features)
- [Installation](#-installation)
- [Requirements](#-requirements)
- [Available Checks](#available-checks)
- [Quick Start](#-quick-start)

  - [1. Using with Pandas](#1-using-with-pandas)
    - [1.1 Duplicate Check](#11-duplicate-check)
    - [1.2 Empty Check](#12-empty-check)
    - [1.3 Unique Check](#13-unique-check)
    - [1.4 Data Type Check](#14-data-type-check)
    - [1.5 String Format (Regex) Check](#15-string-format-regex-check)
    - [1.6 Schema Validation](#16-schema-validation)
    - [1.7 Range Check](#17-range-check)
    - [1.8 Categorical Values Check](#18-categorical-values-check)
    - [1.9 Statistical Distribution Check](#19-statistical-distribution-check)
    - [1.10 Label Balance Check](#110-label-balance-check)
    - [1.11 Data Freshness Check](#111-data-freshness-check)
    - [1.12 Referential Integrity Check](#112-referential-integrity-check)
    - [1.13 Row Count Check](#113-row-count-check)
    - [1.14 Custom Check](#114-custom-check---powerful)
  - [2. Using with Apache Spark](#2-using-with-apache-spark)
    - [2.1 Duplicate Check](#21-duplicate-check)
    - [2.2 Empty Check](#22-empty-check)
    - [2.3 Unique Check](#23-unique-check)
    - [2.4 Data Type Check](#24-data-type-check)
    - [2.5 String Format (Regex) Check](#25-string-format-regex-check)
    - [2.6 Schema Validation](#26-schema-validation)
    - [2.7 Range Check](#27-range-check)
    - [2.8 Categorical Values Check](#28-categorical-values-check)
    - [2.9 Statistical Distribution Check](#29-statistical-distribution-check)
    - [2.10 Label Balance Check](#210-label-balance-check)
    - [2.11 Data Freshness Check](#211-data-freshness-check)
    - [2.12 Referential Integrity Check](#212-referential-integrity-check)
    - [2.13 Row Count Check](#213-row-count-check)
    - [2.14 Custom Check](#214-custom-check---powerful)
  - [3. Using with Apache Flink](#3-using-with-apache-flink)
    - [3.1 Duplicate Check](#31-duplicate-check)
    - [3.2 Empty Check](#32-empty-check)
    - [3.3 Unique Check](#33-unique-check)
    - [3.4 Data Type Check](#34-data-type-check)
    - [3.5 String Format (Regex) Check](#35-string-format-regex-check)
    - [3.6 Schema Validation](#36-schema-validation)
    - [3.7 Range Check](#37-range-check)
    - [3.8 Categorical Values Check](#38-categorical-values-check)
    - [3.9 Statistical Distribution Check](#39-statistical-distribution-check)
    - [3.10 Data Freshness Check](#310-data-freshness-check)
    - [3.11 Referential Integrity Check](#311-referential-integrity-check)
    - [3.13 Row Count Check](#313-row-count-check)
    - [3.14 Custom Check](#314-custom-check---powerful)
  - [4. Using with Ray](#4-using-with-ray)
    - [4.1 Duplicate Check](#41-duplicate-check)
    - [4.2 Empty Check](#42-empty-check)
    - [4.3 Unique Check](#43-unique-check)
    - [4.4 Data Type Check](#44-data-type-check)
    - [4.5 String Format (Regex) Check](#45-string-format-regex-check)
    - [4.6 Schema Validation](#46-schema-validation)
    - [4.7 Range Check](#47-range-check)
    - [4.8 Categorical Values Check](#48-categorical-values-check)
    - [4.9 Statistical Distribution Check](#49-statistical-distribution-check)
    - [4.10 Label Balance Check](#410-label-balance-check)
    - [4.11 Data Freshness Check](#411-data-freshness-check)
    - [4.12 Referential Integrity Check](#412-referential-integrity-check)
    - [4.13 Row Count Check](#413-row-count-check)
    - [4.14 Custom Check](#414-custom-check---powerful)
  - [5. Yaml Config Driven Evaluation for Ray/Spark/Pandas](#5-yaml-config-driven-evaluation-for-sparkraypandas)
    - [API Run Checks from Yaml Config](#api-run_checks_from_yaml)
    - [Example for Ray](#example-yaml-config-dqu_ray_checkyml)
    - [Example for Pandas](#example-for-pandas-python)
    - [Example for PySpark](#example-for-spark-pyspark)
  - [Logging And Monitoring Integration](#-logging--monitoring-integration)
    - [How logging works](#how-logging-works)
    - [What Gets Logged](#what-gets-logged)
    - [Why use loggin?](#why-use-logging)

- [Notes](#-notes)
- [Contributing](#-contributing)
- [License](#-license)

---

## 🔑 Key Features

- ✅ **Multi-Engine Support**: Unified APIs for Pandas, Spark, Flink, and Ray
- 🔍 **Comprehensive Checks**:
  - Null / Empty Checks
  - Duplicate Checks
  - Uniqueness
  - Schema Validation
  - Range Checks
  - String Pattern Checks
  - Categorical Value Checks
  - Data Freshness
  - Referential Integrity
  - Statistical Distribution Checks
  - Custom/Advanced Validations (Powerful to apply any any custom-functions)
- 📊 **Detailed Reporting**:
  - Summary statistics
  - Row-level failed results (optional)
  - Timestamps and run IDs
- 🔌 **Enterprise-Ready**:
  - Designed for large-scale data pipelines
- 🧱 **Extensible**:
  - Easily add new custom rules or checks
  - Plug into existing pipeline frameworks
- 🐍 **Pythonic Interface**:
  - Works natively with Python, PySpark, PyFlink, and Ray APIs

---

## 📦 Installation

```bash
pip install dqu
```

For development:

```bash
git clone https://github.com/keshavksingh/dqu.git
cd dqu
pip install -e .
```

---

## 📋 Requirements

None, make sure the desired environment has in-built support for:

| Dependency   | Version   |
| ------------ | --------- |
| Python       | >= 3.7    |
| pandas       | >= 1.3.0  |
| pyspark      | >= 3.0.0  |
| apache-flink | >= 2.1.0  |
| ray          | >= 2.4.8  |
| scipy        | >= 1.16.1 |

---

## Available Checks

| Check Name               | Class Name                        | Description                                          |
| ------------------------ | --------------------------------- | ---------------------------------------------------- |
| Duplicate Check          | `DquDupCheck`                     | Detects duplicates based on columns                  |
| Empty/Null Check         | `DquEmptyCheck`                   | Checks for missing values                            |
| Uniqueness Check         | `DquUniqueCheck`                  | Ensures column uniqueness                            |
| Data Type Check          | `DquDtypeCheck`                   | Ensures column Data Type                             |
| Row Count Check          | `DquRowCountCheck`                | Row Count Check                                      |
| Range Check              | `DquRangeCheck`                   | Enforces min/max thresholds                          |
| String Format Check      | `DquStringFormatCheck`            | Validates string formats (e.g., email)               |
| Categorical Check        | `DquCategoricalValuesCheck`       | Checks values against an allowed set                 |
| Schema Check             | `DquSchemaValidationCheck`        | Verifies types for each column                       |
| Freshness Check          | `DquDataFreshnessCheck`           | Validates timestamp recency                          |
| Referential Integrity    | `DquReferentialIntegrityCheck`    | Cross-checks between datasets                        |
| Statistical Distribution | `DquStatisticalDistributionCheck` | Validates against mean/std expectations              |
| Custom Python Check      | `DquCustomCheck`                  | Run your own logic for validation (Column/Row) Level |

---

## 🚀 Quick Start

## 1. Using with **Pandas**

You can perform the following validation checks using `PandasEngine`:

### 1.1 Duplicate Check

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.dup_check import DquDupCheck

data = {'id': [1, 2, 2, 4], 'name': ['Alice', 'Bob', 'Bob', 'Dave']}
df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

check = DquDupCheck(dqudf, config={"columns": ["id"]})
result, failed_rows = check.run(evaluation="advanced")

print(result)
print(failed_rows)

{
  "status": "Failed",
  "dqu_check_type": "duplicate_check",
  "dqu_total_count": 4,
  "dqu_failed_count": 2,
  "dqu_passed_count": 2,
  "run_timestamp": "2025-08-05T11:48:43.184619+00:00",
  "run_id": null
}
   id name
   2  Bob
   2  Bob
```

### 1.2 Empty Check

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.empty_check import DquEmptyCheck

data = {'id': [1, 2, 2, 4], 'name': ['', 'Bob', 'Bob', 'Dave']}
df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

check = DquEmptyCheck(dqudf, config={"columns": ["name"]}, run_id="fe6e3991-6e3f-4f53-83cc-b8399bc8fc28")
result, failed_rows = check.run(evaluation="advanced")

print(result)
print(failed_rows)

{
  "status": "Failed",
  "dqu_check_type": "empty_check",
  "dqu_total_count": 4,
  "dqu_failed_count": 1,
  "dqu_passed_count": 3,
  "run_timestamp": "2025-08-05T14:12:27.419489+00:00",
  "run_id": "fe6e3991-6e3f-4f53-83cc-b8399bc8fc28"
}
   id name
   1
```

### 1.3 Unique Check

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.unique_check import DquUniqueCheck

data = {'id': [1, 2, 2, 4], 'name': ['Daniel', 'Bob', 'Bob', 'Dave']}
df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

check = DquUniqueCheck(dqudf, config={"columns": ["id"]}, run_id="fe6e3991-6e3f-4f53-83cc-b8399bc8fc28")
result, failed_rows = check.run(evaluation="advanced")

print(result)
print(failed_rows)

{
  "status": "Failed",
  "dqu_check_type": "unique_check",
  "dqu_total_count": 4,
  "dqu_failed_count": 2,
  "dqu_passed_count": 2,
  "run_timestamp": "2025-08-05T14:13:21.537525+00:00",
  "run_id": "fe6e3991-6e3f-4f53-83cc-b8399bc8fc28"
}
   id name
   2  Bob
   2  Bob
```

### 1.4 Data Type Check

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.dtype_check import DquDtypeCheck

data = {'id': [1, 2, 2, 4], 'name': ['Daniel', 'Bob', 'Bob', 'Dave']}
df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

check = DquDtypeCheck(dqudf, config={"columns": {"id": "int","name": "str"}}, run_id="fe6e3991-6e3f-4f53-83cc-b8399bc8fc28")
result, failed_rows = check.run(evaluation="advanced")

print(result)
print(failed_rows)

{
  "status": "Success",
  "dqu_check_type": "dtype_check",
  "dqu_total_count": 4,
  "dqu_failed_count": 0,
  "dqu_passed_count": 4,
  "run_timestamp": "2025-08-05T14:17:34.098679+00:00",
  "run_id": "fe6e3991-6e3f-4f53-83cc-b8399bc8fc28"
}
Empty DataFrame
Columns: [id, name]
Index: []
```

### 1.5 String Format (Regex) Check

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.stringformat_check import DquStringFormatCheck

data = {
    'id': [1, 2, 3, 4, 5],
    'name': ['Daniel', 'Bob123', 'Alice!', 'Carol', 'Eve']
}
df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

# Check that 'name' column contains only alphabets (A-Z, a-z)
check = DquStringFormatCheck(
    dqudf,
    config={"column": "name", "pattern": r"^[A-Za-z]+$"},
    run_id="fe6e3991-6e3f-4f53-83cc-b8399bc8fc28"
)
result, failed_rows = check.run(evaluation="advanced")

print(result)
print(failed_rows)

{
  "status": "Failed",
  "dqu_check_type": "stringformat_check",
  "dqu_total_count": 5,
  "dqu_failed_count": 2,
  "dqu_passed_count": 3,
  "run_timestamp": "2025-08-05T14:25:34.254124+00:00",
  "run_id": "fe6e3991-6e3f-4f53-83cc-b8399bc8fc28"
}
   id    name
   2  Bob123
   3  Alice!
```

### 1.6 Schema Validation

```python
import pandas as pd
from datetime import datetime
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.schema_check import DquSchemaValidationCheck

data = {
    'id': [1, 2, 3],
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': ['25', '30', '35'],  # age should be int, but it's string here
    'signup_date': ['2023-01-01', '2023-02-15', '2023-03-20']  # should be datetime
}

df = pd.DataFrame(data)
df['signup_date'] = pd.to_datetime(df['signup_date'])
dqudf = DquDataFrame(df)

expected_schema = {
    'id': 'int64',
    'name': 'object',
    'age': 'int64',  # this will fail as actual type is object (string)
    'signup_date': 'datetime64[ns]'  # this should pass
}

check = DquSchemaValidationCheck(
    dqudf,
    config={"expected_schema": expected_schema},
    run_id="a45b0020-f5b0-4a59-b418-0d9f7c3700a6"
)

result = check.run(evaluation="advanced")

print(result)

{
  "status": "Failed",
  "dqu_check_type": "schemavalidation_check",
  "missing_columns": [],
  "type_mismatches": {
    "age": {
      "expected": "int64",
      "actual": "object"
    }
  },
  "dqu_total_count": 3,
  "dqu_failed_count": 1,
  "dqu_passed_count": 3,
  "run_timestamp": "2025-08-05T14:33:41.550181+00:00",
  "run_id": "a45b0020-f5b0-4a59-b418-0d9f7c3700a6"
}
```

### 1.7 Range Check

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.range_check import DquRangeCheck
data = {
    'id': [1, 2, 3, 4],
    'name': ['Alice', 'Bob', 'Charlie', 'Daisy'],
    'age': [22, 17, 65, 80]  # 17 and 80 are outside the valid range
}

df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

check = DquRangeCheck(
    dqudf,
    config={"column": "age", "min": 18, "max": 65},
    run_id="cb376a28-2652-4cb3-a41f-5ec9c0b93dc6"
)

result, failed_rows = check.run(evaluation="advanced")

print(result)
print(failed_rows)

{
  "status": "Failed",
  "dqu_check_type": "range_check",
  "column": "age",
  "range": {
    "min": 18,
    "max": 65
  },
  "dqu_total_count": 4,
  "dqu_failed_count": 2,
  "dqu_passed_count": 2,
  "run_timestamp": "2025-08-05T14:37:09.446949+00:00",
  "run_id": "cb376a28-2652-4cb3-a41f-5ec9c0b93dc6"
}
   id   name  age
   2    Bob   17
   4  Daisy   80
```

### 1.8 Categorical Values Check

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.categoricalvalues_check import DquCategoricalValuesCheck

data = {
    'employee_id': [101, 102, 103, 104],
    'name': ['Alice', 'Bob', 'Charlie', 'Daisy'],
    'department': ['HR', 'Finance', 'Legal', 'Magic']  # 'Magic' is not allowed
}

df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

check = DquCategoricalValuesCheck(
    dqudf,
    config={"column": "department", "allowed_values": ["HR", "Finance", "Legal", "Engineering"]},
    run_id="ec5f4c82-d7a4-43de-9ff5-c18186a684f2"
)

result, failed_rows = check.run(evaluation="advanced")

print(result)
print(failed_rows)

{
  "status": "Failed",
  "dqu_check_type": "categoricalvalues_check",
  "column": "department",
  "allowed_values": [
    "HR",
    "Finance",
    "Legal",
    "Engineering"
  ],
  "dqu_total_count": 4,
  "dqu_failed_count": 1,
  "dqu_passed_count": 3,
  "run_timestamp": "2025-08-05T14:42:07.527741+00:00",
  "run_id": "ec5f4c82-d7a4-43de-9ff5-c18186a684f2"
}
   employee_id   name department
          104  Daisy      Magic
```

### 1.9 Statistical Distribution Check

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.statisticaldistribution_check import DquStatisticalDistributionCheck

data = {
    'employee_id': [1, 2, 3, 4, 5],
    'salary': [70000, 72000, 71000, 69000, 130000]  # 130000 causes drift
}

df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

# Reference stats represent expected distribution
reference_stats = {"mean": 70000, "std": 5000}

check = DquStatisticalDistributionCheck(
    dqudf,
    config={
        "column": "salary",
        "mode": "feature_drift",
        "reference_stats": reference_stats,
        "tolerance": 0.1  # 10% drift tolerance
    },
    run_id="2a9b9370-2b7f-4d01-bd87-57b4c11c86be"
)

result = check.run(evaluation="advanced")
print(result)

{
  "status": "Failed",
  "dqu_check_type": "statisticaldistribution_check",
  "mode": "feature_drift",
  "column": "salary",
  "dqu_drift_mean": 12400.0,
  "dqu_drift_std": 21632.686683847725,
  "dqu_passed": false,
  "run_timestamp": "2025-08-05T14:45:31.369345+00:00",
  "run_id": "2a9b9370-2b7f-4d01-bd87-57b4c11c86be"
}
```

### 1.10 Label Balance Check

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.statisticaldistribution_check import DquStatisticalDistributionCheck

# Imbalanced classification labels
data = {
    'id': [101, 102, 103, 104, 105, 106, 107, 108],
    'label': ['spam', 'spam', 'ham', 'spam', 'spam', 'spam', 'spam', 'ham']
}

df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

check = DquStatisticalDistributionCheck(
    dqudf,
    config={
        "column": "label",
        "mode": "label_balance"
    },
    run_id="c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
)

result = check.run(evaluation="advanced")

print(result)

{
  "status": "Success",
  "dqu_check_type": "statisticaldistribution_check",
  "mode": "label_balance",
  "column": "label",
  "dqu_distribution": {
    "spam": 0.75,
    "ham": 0.25
  },
  "dqu_passed": true,
  "run_timestamp": "2025-08-05T14:49:13.961807+00:00",
  "run_id": "c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
}


import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.statisticaldistribution_check import DquStatisticalDistributionCheck

# Extremely imbalanced classification labels
data = {
    'id': list(range(1, 21)),
    'label': ['spam'] * 19 + ['ham']  # 95% spam, 5% ham
}

df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

check = DquStatisticalDistributionCheck(
    dqudf,
    config={
        "column": "label",
        "mode": "label_balance"
    },
    run_id="c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
)

result = check.run(evaluation="advanced")

print(result)

{
  "status": "Failed",
  "dqu_check_type": "statisticaldistribution_check",
  "mode": "label_balance",
  "column": "label",
  "dqu_distribution": {
    "spam": 0.95,
    "ham": 0.05
  },
  "dqu_passed": false,
  "run_timestamp": "2025-08-05T14:53:30.108379+00:00",
  "run_id": "c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
}
```

### 1.11 Data Freshness Check

```python
import pandas as pd
from datetime import datetime, timedelta
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.datafreshness_check import DquDataFreshnessCheck

# Create a dataframe with timestamps (milliseconds since epoch)
now = datetime.utcnow()
data = {
    "id": [1, 2, 3, 4],
    "created_at": [now,now,now,now]
}

df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

check = DquDataFreshnessCheck(
    dqudf,
    config={
        "column": "created_at",
        "freshness_threshold": "2d"
    },
    run_id="c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
)

result = check.run(evaluation="advanced")

print(result)

{
  "status": "Success",
  "dqu_check_type": "datafreshness_check",
  "column": "created_at",
  "latest_timestamp": "2025-08-05 14:59:41.348080",
  "cutoff_timestamp": "2025-08-03 14:59:41.350868",
  "dqu_passed": true,
  "run_timestamp": "2025-08-05T14:59:41.350955+00:00",
  "run_id": "c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
}
```

### 1.12 Referential Integrity Check

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.referential_integrity_check import DquReferentialIntegrityCheck

# Main DataFrame (e.g., transactions)
main_data = {
    "user_id": [101, 102, 103, 104],
    "amount": [50.0, 75.5, 23.0, 99.9]
}
main_df = pd.DataFrame(main_data)
dqu_main = DquDataFrame(main_df)

# Reference DataFrame (e.g., users table)
ref_data = {
    "id": [101, 102, 103, 104]  # All user_ids from main_df exist here
}
ref_df = pd.DataFrame(ref_data)
dqu_ref = DquDataFrame(ref_df)

check = DquReferentialIntegrityCheck(
    dqu_main,
    config={
        "column": "user_id",
        "reference_df": dqu_ref.df,
        "reference_column": "id"
    },
    run_id="c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
)

result, failed_rows = check.run(evaluation="advanced")

print(result)
print(failed_rows)

{
  "status": "Success",
  "dqu_check_type": "referentialintegrity_check",
  "dqu_total_count": 4,
  "dqu_failed_count": 0,
  "dqu_passed_count": 4,
  "run_timestamp": "2025-08-05T15:05:04.944163+00:00",
  "run_id": "c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
}
Empty DataFrame
Columns: [user_id, amount]
Index: []

import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.referential_integrity_check import DquReferentialIntegrityCheck

# Main DataFrame (e.g., transactions)
main_data = {
    "user_id": [101, 102, 103, 104],
    "amount": [50.0, 75.5, 23.0, 99.9]
}
main_df = pd.DataFrame(main_data)
dqu_main = DquDataFrame(main_df)

# Reference DataFrame (e.g., users table)
ref_data = {
    "id": [101, 102, 103, 107]  # All user_ids from main_df exist here 107 doesn't
}
ref_df = pd.DataFrame(ref_data)
dqu_ref = DquDataFrame(ref_df)

check = DquReferentialIntegrityCheck(
    dqu_main,
    config={
        "column": "user_id",
        "reference_df": dqu_ref.df,
        "reference_column": "id"
    },
    run_id="c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
)

result, failed_rows = check.run(evaluation="advanced")

print(result)
print(failed_rows)

{
  "status": "Failed",
  "dqu_check_type": "referentialintegrity_check",
  "dqu_total_count": 4,
  "dqu_failed_count": 1,
  "dqu_passed_count": 3,
  "run_timestamp": "2025-08-05T15:05:58.191848+00:00",
  "run_id": "c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
}
   user_id  amount
      104    99.9
```

### 1.13 Row Count Check

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.rowcount_check import DquRowCountCheck

data = {
    "id": [1, 2, 3, 4, 5],
    "name": ["Alice", "Bob", "Charlie", "David", "Eva"]
}

df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

# Check that the row count is between 3 and 10
check = DquRowCountCheck(
    dqudf,
    config={
        "min": 3,
        "max": 10
    },
    run_id="123e4567-e89b-12d3-a456-426614174000"
)

result = check.run(evaluation="advanced")

print(result)

{
  "status": "Success",
  "dqu_check_type": "rowcount_check",
  "dqu_total_count": 5,
  "min_required": 3,
  "max_allowed": 10,
  "run_timestamp": "2025-08-05T15:11:19.547228+00:00",
  "run_id": "123e4567-e89b-12d3-a456-426614174000"
}

import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.rowcount_check import DquRowCountCheck

data = {
    "id": [1, 2, 3, 4, 5],
    "name": ["Alice", "Bob", "Charlie", "David", "Eva"]
}

df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

# Check that the row count is between 3 and 10
check = DquRowCountCheck(
    dqudf,
    config={
        "min": 30,
        "max": 100
    },
    run_id="123e4567-e89b-12d3-a456-426614174000"
)

result = check.run(evaluation="advanced")

print(result)

{
  "status": "Failed",
  "dqu_check_type": "rowcount_check",
  "dqu_total_count": 5,
  "min_required": 30,
  "max_allowed": 100,
  "run_timestamp": "2025-08-05T15:12:09.836859+00:00",
  "run_id": "123e4567-e89b-12d3-a456-426614174000"
}
```

### 1.14 Custom Check - (\*\*Powerful)

```python
import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.custom_check import DquCustomCheck

data = {
    "id": [1, 2, 3, 4],
    "score": [85, 92, -10, 74]
}

df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

# Run a custom check: score should be >= 0
check = DquCustomCheck(
    dqudf,
    config={
        "column": "score",
        "func": lambda x: x >= 0
    },
    run_id="123e4567-e89b-12d3-a456-426614174000"
)

result, failed_rows = check.run(evaluation="advanced")

print(result)
print(failed_rows)

{
  "status": "Failed",
  "dqu_check_type": "custom_check_column",
  "column": "score",
  "dqu_total_count": 4,
  "dqu_failed_count": 1,
  "dqu_passed_count": 3,
  "run_timestamp": "2025-08-05T15:16:55.688547+00:00",
  "run_id": "123e4567-e89b-12d3-a456-426614174000"
}

import pandas as pd
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.custom_check import DquCustomCheck

data = {
    "id": [1, 2, 3, 4],
    "age": [25, 17, 35, 45],
    "country": ["US", "US", "CA", "IN"]
}

df = pd.DataFrame(data)
dqudf = DquDataFrame(df)

# Run a custom row-level check: age >= 18 and country is 'US'
check = DquCustomCheck(
    dqudf,
    config={
        "func": lambda row: row["age"] >= 18 and row["country"] == "US"
    },
    run_id="123e4567-e89b-12d3-a456-426614174000"
)

result, failed_rows = check.run(evaluation="advanced")

print(result)
print(failed_rows)

{
  "status": "Failed",
  "dqu_check_type": "custom_check_row",
  "column": null,
  "dqu_total_count": 4,
  "dqu_failed_count": 3,
  "dqu_passed_count": 1,
  "run_timestamp": "2025-08-05T15:19:25.942403+00:00",
  "run_id": "123e4567-e89b-12d3-a456-426614174000"
}
   id  age country
   2   17      US
   3   35      CA
   4   45      IN
```

---

## 📌 Notes

- All checks support two `evaluation` modes: `"basic"` (summary JSON) and `"advanced"` (summary JSON + failed rows).
- Optiinally Ensure `run_id` is set for tracking results consistently across runs.

## 2. Using with **Apache Spark**

You can perform the following validation checks using `SparkEngine`:

### 2.1 Duplicate Check

```python
from pyspark.sql import SparkSession
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.dup_check import DquDupCheck

spark = SparkSession.builder.appName("dqu-example").getOrCreate()
df = spark.createDataFrame([(1, "Alice"), (2, "Bob"), (2, "Bob"), (3, "Dave")], schema=["id", "name"])
dqudf = DquDataFrame(df)

check = DquDupCheck(dqudf, config={"columns": ["id"]}, run_id="fe6e3991-6e3f-4f53-83cc-b8399bc8fc28")
result, failed_rows = check.run(evaluation="advanced")

print(result)
failed_rows.show()

{
  "status": "Failed",
  "dqu_check_type": "duplicate_check",
  "dqu_total_count": 4,
  "dqu_failed_count": 2,
  "dqu_passed_count": 2,
  "run_timestamp": "2025-08-06T04:58:52.096909+00:00",
  "run_id": "fe6e3991-6e3f-4f53-83cc-b8399bc8fc28"
}
+---+----+-----+
| id|name|count|
+---+----+-----+
|  2| Bob|    2|
|  2| Bob|    2|
+---+----+-----+
```

### 2.2 Empty Check

```python
from pyspark.sql import SparkSession
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.empty_check import DquEmptyCheck

spark = SparkSession.builder.appName("dqu-example").getOrCreate()
df = spark.createDataFrame([(1, "A"), (None, "B")], schema=["id", "category"])
dqudf = DquDataFrame(df)
check = DquEmptyCheck(dqudf, config={"columns": ["id"]}, run_id="fe6e3991-6e3f-4f53-83cc-b8399bc8fc28")

result, failed = check.run(evaluation="advanced")
print(result)
failed.show()

{
  "status": "Failed",
  "dqu_check_type": "empty_check",
  "dqu_total_count": 2,
  "dqu_failed_count": 1,
  "dqu_passed_count": 1,
  "run_timestamp": "2025-08-06T04:57:13.465760+00:00",
  "run_id": "fe6e3991-6e3f-4f53-83cc-b8399bc8fc28"
}
+----+--------+
|  id|category|
+----+--------+
|NULL|       B|
+----+--------+
```

### 2.3 Unique Check

```python
from pyspark.sql import SparkSession
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.unique_check import DquUniqueCheck

spark = SparkSession.builder.appName("dqu-example").getOrCreate()
df = spark.createDataFrame([(1, "Daniel"),(2, "Bob"),(2, "Bob"),(3, "Dave")], schema=["id", "name"])
dqudf = DquDataFrame(df)

check = DquUniqueCheck(dqudf, config={"columns": ["id"]}, run_id="fe6e3991-6e3f-4f53-83cc-b8399bc8fc28")
result, failed_rows = check.run(evaluation="advanced")

print(result)
failed_rows.show()

{
  "status": "Failed",
  "dqu_check_type": "unique_check",
  "dqu_total_count": 4,
  "dqu_failed_count": 2,
  "dqu_passed_count": 2,
  "run_timestamp": "2025-08-06T05:36:24.368083+00:00",
  "run_id": "fe6e3991-6e3f-4f53-83cc-b8399bc8fc28"
}
+---+----+-----+
| id|name|count|
+---+----+-----+
|  2| Bob|    2|
|  2| Bob|    2|
+---+----+-----+
```

### 2.4 Data Type Check

```python
from pyspark.sql import SparkSession
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.dtype_check import DquDtypeCheck

spark = SparkSession.builder.appName("dqu-example").getOrCreate()
df = spark.createDataFrame([(1, "Daniel"),(2, "Bob"),(2, "Bob"),(3, "Dave")], schema=["id", "name"])
dqudf = DquDataFrame(df)

check = DquDtypeCheck(dqudf, config={"columns": {"id": "int","name": "str"}}, run_id="fe6e3991-6e3f-4f53-83cc-b8399bc8fc28")
result, failed_rows = check.run(evaluation="advanced")

print(result)
failed_rows.show()

{
  "status": "Success",
  "dqu_check_type": "dtype_check",
  "dqu_total_count": 4,
  "dqu_failed_count": 0,
  "dqu_passed_count": 4,
  "run_timestamp": "2025-08-06T05:37:39.660593+00:00",
  "run_id": "fe6e3991-6e3f-4f53-83cc-b8399bc8fc28"
}
+---+----+
| id|name|
+---+----+
+---+----+
```

### 2.5 String Format (Regex) Check

```python
from pyspark.sql import SparkSession
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.stringformat_check import DquStringFormatCheck

spark = SparkSession.builder.appName("dqu-example").getOrCreate()
df = spark.createDataFrame([(1, "Daniel"),(2, "Bob123"),(3, "Alice!"),(4, "Dave")], schema=["id", "name"])
dqudf = DquDataFrame(df)

# Check that 'name' column contains only alphabets (A-Z, a-z)
check = DquStringFormatCheck(
    dqudf,
    config={"column": "name", "pattern": r"^[A-Za-z]+$"},
    run_id="fe6e3991-6e3f-4f53-83cc-b8399bc8fc28"
)
result, failed_rows = check.run(evaluation="advanced")

print(result)
failed_rows.show()


{
  "status": "Failed",
  "dqu_check_type": "stringformat_check",
  "dqu_total_count": 4,
  "dqu_failed_count": 2,
  "dqu_passed_count": 2,
  "run_timestamp": "2025-08-06T05:38:48.615612+00:00",
  "run_id": "fe6e3991-6e3f-4f53-83cc-b8399bc8fc28"
}
+---+------+
| id|  name|
+---+------+
|  2|Bob123|
|  3|Alice!|
+---+------+
```

### 2.6 Schema Validation

```python
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType
from datetime import datetime
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.schema_check import DquSchemaValidationCheck

spark = SparkSession.builder.appName("DQU_SchemaValidation_Spark").getOrCreate()
data = [
    (1, "Alice", "25", datetime(2023, 1, 1)),
    (2, "Bob", "30", datetime(2023, 2, 15)),
    (3, "Charlie", "35", datetime(2023, 3, 20))
]

schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("age", StringType(), True),  # Incorrect type (should be IntegerType)
    StructField("signup_date", TimestampType(), True)
])

df = spark.createDataFrame(data, schema)
dqudf = DquDataFrame(df)

# Expected schema
expected_schema = {
    "id": "int",
    "name": "string",
    "age": "int",           # Intentional mismatch
    "signup_date": "timestamp"
}

check = DquSchemaValidationCheck(
    dqudf,
    config={"expected_schema": expected_schema},
    run_id="a45b0020-f5b0-4a59-b418-0d9f7c3700a6"
)

result = check.run(evaluation="advanced")
print(result)


{
  "status": "Failed",
  "dqu_check_type": "schemavalidation_check",
  "missing_columns": [],
  "type_mismatches": {
    "age": {
      "expected": "int",
      "actual": "string"
    }
  },
  "dqu_total_count": 3,
  "dqu_failed_count": 1,
  "dqu_passed_count": 3,
  "run_timestamp": "2025-08-06T05:45:01.127153+00:00",
  "run_id": "a45b0020-f5b0-4a59-b418-0d9f7c3700a6"
}
```

### 2.7 Range Check

```python
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.range_check import DquRangeCheck

spark = SparkSession.builder.appName("DQU_RangeCheck_Spark").getOrCreate()
data = [
    (1, 'Alice', 22),
    (2, 'Bob', 17),       # Out of range
    (3, 'Charlie', 65),
    (4, 'Daisy', 80)      # Out of range
]

schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True)
])
df = spark.createDataFrame(data, schema)
dqudf = DquDataFrame(df)

# Run Range Check (valid age: 18 to 65)
check = DquRangeCheck(
    dqudf,
    config={"column": "age", "min": 18, "max": 65},
    run_id="cb376a28-2652-4cb3-a41f-5ec9c0b93dc6"
)
result, failed_rows = check.run(evaluation="advanced")
print(result)
failed_rows.show()

{
  "status": "Failed",
  "dqu_check_type": "range_check",
  "column": "age",
  "range": {
    "min": 18,
    "max": 65
  },
  "dqu_total_count": 4,
  "dqu_failed_count": 2,
  "dqu_passed_count": 2,
  "run_timestamp": "2025-08-06T05:48:31.085152+00:00",
  "run_id": "cb376a28-2652-4cb3-a41f-5ec9c0b93dc6"
}
+---+-----+---+
| id| name|age|
+---+-----+---+
|  2|  Bob| 17|
|  4|Daisy| 80|
+---+-----+---+
```

### 2.8 Categorical Values Check

```python
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.categoricalvalues_check import DquCategoricalValuesCheck

spark = SparkSession.builder.appName("DQU_CategoricalValuesCheck_Spark").getOrCreate()
data = [
    (101, 'Alice', 'HR'),
    (102, 'Bob', 'Finance'),
    (103, 'Charlie', 'Legal'),
    (104, 'Daisy', 'Magic')  # Invalid category
]

schema = StructType([
    StructField("employee_id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("department", StringType(), True)
])

df = spark.createDataFrame(data, schema)
dqudf = DquDataFrame(df)
check = DquCategoricalValuesCheck(
    dqudf,
    config={
        "column": "department",
        "allowed_values": ["HR", "Finance", "Legal", "Engineering"]
    },
    run_id="ec5f4c82-d7a4-43de-9ff5-c18186a684f2"
)
result, failed_rows = check.run(evaluation="advanced")
print(result)
failed_rows.show()

{
  "status": "Failed",
  "dqu_check_type": "categoricalvalues_check",
  "column": "department",
  "allowed_values": [
    "HR",
    "Finance",
    "Legal",
    "Engineering"
  ],
  "dqu_total_count": 4,
  "dqu_failed_count": 1,
  "dqu_passed_count": 3,
  "run_timestamp": "2025-08-06T05:51:05.451955+00:00",
  "run_id": "ec5f4c82-d7a4-43de-9ff5-c18186a684f2"
}
+-----------+-----+----------+
|employee_id| name|department|
+-----------+-----+----------+
|        104|Daisy|     Magic|
+-----------+-----+----------+

```

### 2.9 Statistical Distribution Check

```python
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, FloatType
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.statisticaldistribution_check import DquStatisticalDistributionCheck

spark = SparkSession.builder.appName("DQU_FeatureDriftCheck_Spark").getOrCreate()
data = [
    (1, 70000.0),
    (2, 72000.0),
    (3, 71000.0),
    (4, 69000.0),
    (5, 130000.0)  # drift
]

schema = StructType([
    StructField("employee_id", IntegerType(), True),
    StructField("salary", FloatType(), True)
])

df = spark.createDataFrame(data, schema)
dqudf = DquDataFrame(df)

# Reference statistics and drift config
reference_stats = {"mean": 70000, "std": 5000}

check = DquStatisticalDistributionCheck(
    dqudf,
    config={
        "column": "salary",
        "mode": "feature_drift",
        "reference_stats": reference_stats,
        "tolerance": 0.1  # 10% tolerance
    },
    run_id="2a9b9370-2b7f-4d01-bd87-57b4c11c86be"
)
result = check.run(evaluation="advanced")
print(result)


{
  "status": "Failed",
  "dqu_check_type": "statisticaldistribution_check",
  "mode": "feature_drift",
  "column": "salary",
  "dqu_drift_mean": 12400.0,
  "dqu_drift_std": 21632.686683847725,
  "dqu_passed": false,
  "run_timestamp": "2025-08-06T05:54:04.584192+00:00",
  "run_id": "2a9b9370-2b7f-4d01-bd87-57b4c11c86be"
}
```

### 2.10 Label Balance Check

```python
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.statisticaldistribution_check import DquStatisticalDistributionCheck

spark = SparkSession.builder.appName("DQU_LabelBalanceCheck_Spark").getOrCreate()

# Sample imbalanced label data
data = [
    (101, 'spam'),
    (102, 'spam'),
    (103, 'ham'),
    (104, 'spam'),
    (105, 'spam'),
    (106, 'spam'),
    (107, 'spam'),
    (108, 'ham')
]

schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("label", StringType(), True)
])

df = spark.createDataFrame(data, schema)
dqudf = DquDataFrame(df)
check = DquStatisticalDistributionCheck(
    dqudf,
    config={
        "column": "label",
        "mode": "label_balance"
    },
    run_id="c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
)

result = check.run(evaluation="advanced")
print(result)

{
  "status": "Success",
  "dqu_check_type": "statisticaldistribution_check",
  "mode": "label_balance",
  "column": "label",
  "dqu_distribution": {
    "spam": 0.75,
    "ham": 0.25
  },
  "dqu_passed": true,
  "run_timestamp": "2025-08-06T05:57:09.014074+00:00",
  "run_id": "c8395aa5-ecf0-40c0-9cc9-0a99e22f73a9"
}


from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from dqu.kernel.dataframe import DquDataFrame
from dqu.validators.statisticaldistribution_check import DquStatisticalDistributionCheck

spark = SparkSession.builder.appName("DQU_LabelBalanceFail_Spark").getOrCreate()

# Prepare data (95% spam, 5% ham)
data = [(i, 'spam') for i in range(1, 20)] + [(20, 'ham')]

schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("labe