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

┌──────────────────────┐
│ 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:

pip install spark-data-quality

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

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

Verify the installation:

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

---

Quick Start

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.

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.

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:

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.

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.

{
    "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

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

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

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.

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.

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:

North
South
East
West

Expected Daily Load Volume

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

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

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

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:

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.

pip install "great-expectations==0.18.12"

Spark or Java Initialization Errors

Verify:

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

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