Metadata-Version: 2.4
Name: spark-data-quality
Version: 1.1.1
Summary: SparkDQAgent — Data Quality validation package for K8s Spark pods
Author-email: khailas <khailas.rangath@saal.ai>
License-Expression: MIT
Classifier: Programming Language :: Python :: 3
Classifier: Operating System :: OS Independent
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: requests>=2.28.0
Requires-Dist: great-expectations==0.18.12
Requires-Dist: trino>=0.320.0
Requires-Dist: PyYAML>=6.0
Provides-Extra: spark
Requires-Dist: pyspark>=3.1.1; extra == "spark"
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Dynamic: license-file

# spark-data-quality

A lightweight Spark data quality agent that retrieves centrally managed validation rules from a Data Quality Engine, executes them against a PySpark DataFrame using Great Expectations, and publishes the validation results back to the platform.

## Overview

`spark-data-quality` is intended for Spark-based data pipelines running in Kubernetes, scheduled jobs, notebooks, or other PySpark environments.

Instead of embedding data quality rules directly inside every pipeline, teams can define and manage those rules centrally in the Data Quality Engine. At runtime, the package:

1. Identifies the dataset using its catalog, schema, and table name.
2. Retrieves the configured data quality assertions.
3. Executes the assertions against a Spark DataFrame.
4. Sends the results back to the Data Quality Engine.
5. Returns a compact execution summary to the calling application.

This helps data engineering teams apply consistent validation rules across pipelines while retaining centralized monitoring and governance.

---

## Key Features

- Native validation of PySpark DataFrames.
- Centrally managed data quality assertions.
- Great Expectations-based rule execution.
- Support for common completeness, uniqueness, validity, volume, and statistical checks.
- Parallel execution of multiple assertions.
- Validation results published to the Data Quality Engine.
- Simple API designed for batch pipelines and Kubernetes-hosted Spark workloads.

---

## How It Works

```text
┌──────────────────────┐
│ Data Quality Engine  │
│                      │
│ Stores validation    │
│ rules and results    │
└──────────┬───────────┘
           │
           │ 1. Fetch assertions
           ▼
┌──────────────────────┐
│ SparkDQAgent         │
│                      │
│ Executes assertions  │
│ using Great          │
│ Expectations         │
└──────────┬───────────┘
           │
           │ 2. Validate
           ▼
┌──────────────────────┐
│ PySpark DataFrame    │
└──────────┬───────────┘
           │
           │ 3. Publish results
           ▼
┌──────────────────────┐
│ Data Quality Engine  │
└──────────────────────┘
```

---

## Requirements

- Python 3.8 or later.
- PySpark 3.1.1 or later.
- Great Expectations 0.18.12.
- Access to a running Data Quality Engine instance.
- Data quality assertions configured for the target dataset.
- Java and Spark configured in the execution environment.

For Trino-backed datasets, the runtime must also have access to the appropriate Trino JDBC driver when the DataFrame is loaded through JDBC.

---

## Installation

Install the package from PyPI:

```bash
pip install spark-data-quality
```

For an environment where Spark dependencies are not already installed, use the package's Spark extra when applicable:

```bash
pip install "spark-data-quality[spark]"
```

Verify the installation:

```bash
python -c "from spark_dq.quality import SparkDQAgent; print('spark-data-quality installed')"
```

---

## Quick Start

```python
from pyspark.sql import SparkSession
from spark_dq.quality import SparkDQAgent

spark = (
    SparkSession.builder
    .appName("sales-data-quality")
    .master("local[*]")
    .getOrCreate()
)

agent = SparkDQAgent(
    catalog="mycatalog",
    schema="myschema",
    table="sales_data",
    data_quality_url="https://dq.example.com/api/v1/spark",
    catalog_type="unmanaged",
    trino_host="trino.example.com:443",
    trino_user="service_account",
    trino_pwd="your_password",
)

df = (
    spark.read
    .format("jdbc")
    .option("url", "jdbc:trino://trino.example.com:443?SSL=true")
    .option("driver", "io.trino.jdbc.TrinoDriver")
    .option("user", "service_account")
    .option("password", "your_password")
    .option(
        "query",
        "SELECT * FROM mycatalog.myschema.sales_data",
    )
    .load()
)

results = agent.execute_data_quality(df)

for suite_name, summary in results.items():
    print(
        f"{suite_name}: "
        f"{summary['successful_expectations']}/"
        f"{summary['evaluated_expectations']} checks passed "
        f"({summary['success_percent']}%)"
    )

spark.stop()
```

---

## Configuration

Create a `SparkDQAgent` for the dataset that will be validated.

```python
agent = SparkDQAgent(
    catalog="mycatalog",
    schema="myschema",
    table="sales_data",
    data_quality_url="https://dq.example.com/api/v1/spark",
    catalog_type="unmanaged",
    trino_host="trino.example.com:443",
    trino_user="service_account",
    trino_pwd="your_password",
)
```

| Parameter | Description | Example |
|---|---|---|
| `catalog` | Catalog containing the target dataset. | `mycatalog` |
| `schema` | Schema containing the target table. | `myschema` |
| `table` | Table associated with the configured quality rules. | `sales_data` |
| `data_quality_url` | Base API endpoint of the Data Quality Engine Spark integration. | `https://dq.example.com/api/v1/spark` |
| `catalog_type` | Catalog integration type used by the Data Quality Engine. | `unmanaged` |
| `trino_host` | Trino endpoint used for dataset connectivity or metadata operations. | `trino.example.com:443` |
| `trino_user` | Trino service account or username. | `service_account` |
| `trino_pwd` | Trino password or secret. | `********` |

### Recommended Secret Handling

Do not hard-code passwords in source code. Load credentials from environment variables or a secret manager.

```python
import os

agent = SparkDQAgent(
    catalog=os.environ["DQ_CATALOG"],
    schema=os.environ["DQ_SCHEMA"],
    table=os.environ["DQ_TABLE"],
    data_quality_url=os.environ["DQ_ENGINE_URL"],
    catalog_type=os.getenv("DQ_CATALOG_TYPE", "unmanaged"),
    trino_host=os.environ["TRINO_HOST"],
    trino_user=os.environ["TRINO_USER"],
    trino_pwd=os.environ["TRINO_PASSWORD"],
)
```

Example environment variables:

```bash
export DQ_CATALOG=mycatalog
export DQ_SCHEMA=myschema
export DQ_TABLE=sales_data
export DQ_ENGINE_URL=https://dq.example.com/api/v1/spark
export DQ_CATALOG_TYPE=unmanaged
export TRINO_HOST=trino.example.com:443
export TRINO_USER=service_account
export TRINO_PASSWORD='replace-with-secret'
```

---

## API Reference

### `SparkDQAgent.execute_data_quality(df)`

Executes all configured data quality assertions for the agent's catalog, schema, and table against the supplied PySpark DataFrame.

```python
results = agent.execute_data_quality(df)
```

#### Parameters

| Parameter | Type | Description |
|---|---|---|
| `df` | `pyspark.sql.DataFrame` | Spark DataFrame to validate. |

#### Processing Flow

The method performs the following operations:

1. Retrieves assertions configured for the target dataset.
2. Converts the configured assertions into executable validation checks.
3. Evaluates the checks against the DataFrame.
4. Publishes individual validation outcomes to the Data Quality Engine.
5. Returns an aggregated summary grouped by validation suite.

#### Return Value

A dictionary containing one or more suite-level summaries.

```python
{
    "sales_data_93": {
        "evaluated_expectations": 5,
        "successful_expectations": 4,
        "unsuccessful_expectations": 1,
        "success_percent": 80.0,
    }
}
```

| Field | Description |
|---|---|
| `evaluated_expectations` | Total number of assertions evaluated. |
| `successful_expectations` | Number of assertions that passed. |
| `unsuccessful_expectations` | Number of assertions that failed. |
| `success_percent` | Percentage of successful assertions. |

---

## Using Results in a Pipeline

### Fail the Pipeline When Any Check Fails

```python
results = agent.execute_data_quality(df)

failed_checks = sum(
    suite["unsuccessful_expectations"]
    for suite in results.values()
)

if failed_checks > 0:
    raise RuntimeError(
        f"Data quality validation failed: {failed_checks} check(s) failed"
    )
```

### Enforce a Minimum Quality Threshold

```python
minimum_pass_rate = 95.0
results = agent.execute_data_quality(df)

below_threshold = {
    suite_name: suite["success_percent"]
    for suite_name, suite in results.items()
    if suite["success_percent"] < minimum_pass_rate
}

if below_threshold:
    raise RuntimeError(
        f"Data quality pass rate is below {minimum_pass_rate}%: "
        f"{below_threshold}"
    )
```

### Log a Compact Summary

```python
import logging

logger = logging.getLogger(__name__)

results = agent.execute_data_quality(df)

for suite_name, summary in results.items():
    logger.info(
        "DQ suite=%s evaluated=%s passed=%s failed=%s pass_rate=%s%%",
        suite_name,
        summary["evaluated_expectations"],
        summary["successful_expectations"],
        summary["unsuccessful_expectations"],
        summary["success_percent"],
    )
```

---

## Supported Assertion Examples

The package can execute assertions configured in the Data Quality Engine, including common Great Expectations assertion types such as:

| Data Quality Dimension | Assertion Type | Typical Use |
|---|---|---|
| Completeness | `expect_column_values_to_not_be_null` | Ensure mandatory fields are populated. |
| Uniqueness | `expect_column_values_to_be_unique` | Detect duplicate identifiers. |
| Validity | `expect_column_values_to_be_between` | Validate numeric or date ranges. |
| Volume | `expect_table_row_count_to_be_between` | Detect missing or unexpectedly large loads. |
| Cardinality | `expect_column_unique_value_count_to_be_between` | Validate the expected number of distinct values. |
| Distribution | `expect_column_median_to_be_between` | Detect unusual shifts in numeric data. |
| Format | `expect_column_values_to_match_regex` | Validate codes, emails, identifiers, or patterns. |
| Length | `expect_column_value_lengths_to_be_between` | Validate minimum and maximum text lengths. |

The exact assertions available to a pipeline depend on the rules configured in the Data Quality Engine.

---

## Example Validation Scenarios

### Mandatory Customer Identifier

Business rule: every record must contain a customer identifier.

```text
Assertion: expect_column_values_to_not_be_null
Column: customer_id
```

### Valid Transaction Amount

Business rule: transaction amounts must be positive and must not exceed the agreed operational limit.

```text
Assertion: expect_column_values_to_be_between
Column: amount
Minimum: 1
Maximum: 500000
```

### Expected Regional Values

Business rule: region values must conform to the approved domain.

A suitable configured assertion can validate values such as:

```text
North
South
East
West
```

### Expected Daily Load Volume

Business rule: a daily pipeline should load between 95,000 and 105,000 records.

```text
Assertion: expect_table_row_count_to_be_between
Minimum: 95000
Maximum: 105000
```

---

## Kubernetes and Spark Job Usage

The package can be included in a Spark container image or installed when the job starts.

### Dockerfile Example

```dockerfile
FROM apache/spark-py:3.5.1

USER root

RUN pip install --no-cache-dir spark-data-quality

USER 185

COPY jobs /opt/spark/jobs
```

### `spark-submit` Example

```bash
spark-submit \
  --master k8s://https://kubernetes.default.svc \
  --deploy-mode cluster \
  --name sales-data-quality \
  --conf spark.kubernetes.container.image=registry.example.com/spark-dq:1.0.0 \
  --conf spark.executor.instances=4 \
  local:///opt/spark/jobs/validate_sales.py
```

Provide secrets through Kubernetes Secrets, workload identity, Vault integration, or another approved secret-management mechanism rather than command-line arguments.

---

## Operational Recommendations

- Validate data immediately after ingestion and before publishing it to curated or consumption layers.
- Treat critical-rule failures differently from informational warnings at the orchestration layer.
- Use a dedicated service account for the Data Quality Engine and Trino.
- Configure TLS for all Data Quality Engine and Trino communication.
- Avoid collecting or logging raw sensitive values as part of validation failure messages.
- Pin package versions in production deployments.
- Test new rule configurations in a non-production environment before rollout.
- Monitor validation duration as the number of assertions and dataset volume increase.

Example pinned dependency:

```text
spark-data-quality==1.1.0
```

---

## Troubleshooting

### No Assertions Are Executed

Check that:

- The catalog, schema, and table values exactly match the dataset configured in the Data Quality Engine.
- Assertions have been created and enabled for the dataset.
- The Spark job can reach the Data Quality Engine endpoint.
- Required authentication and network policies are configured.

### Unable to Connect to Trino

Check that:

- The Trino hostname and port are correct.
- TLS settings match the target environment.
- The service account has permission to access the catalog and schema.
- The Trino JDBC driver is available to Spark when JDBC loading is used.
- Kubernetes NetworkPolicies, firewall rules, proxies, and DNS resolution allow the connection.

### Great Expectations Compatibility Errors

The current package documentation specifies Great Expectations `0.18.12`. Avoid upgrading Great Expectations independently without compatibility testing.

```bash
pip install "great-expectations==0.18.12"
```

### Spark or Java Initialization Errors

Verify:

```bash
python --version
java -version
spark-submit --version
```

Ensure `JAVA_HOME` and Spark-related environment variables are correctly configured.

### Data Quality Results Are Not Visible in the Platform

Check that:

- The Data Quality Engine API endpoint is correct.
- The Spark job has outbound connectivity to the API.
- API requests are not blocked by a proxy or certificate-validation issue.
- The dataset identifier used by the agent matches the platform configuration.

---

## Version Compatibility

| Component | Supported Version |
|---|---|
| Python | 3.8+ |
| PySpark | 3.1.1+ |
| Great Expectations | 0.18.12 |
| Package | 1.1.0 |

Compatibility should be verified against the Spark, Java, and Kubernetes versions used by your deployment environment.

---

## Security Considerations

- Never commit Trino passwords or API credentials to source control.
- Use environment variables only when a stronger secret store is unavailable.
- Prefer Kubernetes Secrets, HashiCorp Vault, Azure Key Vault, AWS Secrets Manager, or an equivalent managed service.
- Restrict service accounts to the minimum required catalog, schema, and API permissions.
- Use encrypted connections for both Trino and the Data Quality Engine.
- Rotate credentials in accordance with organizational security policies.

---

## License

This project is licensed under the MIT License.

---

## Package Summary

```text
Package: spark-data-quality
Primary class: spark_dq.quality.SparkDQAgent
Primary method: execute_data_quality(df)
Purpose: Execute centrally configured data quality rules against PySpark DataFrames
Validation engine: Great Expectations
Minimum Python version: 3.8
```
