15. Data Quality in PySpark

Six Quality Dimensions Wheel

🏗️ Data Quality Architecture

📚 Detailed Explanation

Why Data Quality Matters

• Poor data quality → incorrect business decisions, failed processes, loss of trust
• Requires validation rules, anomaly detection, and continuous monitoring
• Must be embedded at every stage of the data pipeline

Six Quality Dimensions

Key Takeaway: A comprehensive quality framework evaluates all six dimensions and provides a holistic quality score.

Validation Rules

Implementation options:

• DataFrame operations
• SQL constraints
• External libraries (Great Expectations, Deequ)

Anomaly Detection

• Identifies data points deviating significantly from expected patterns
• Statistical methods:
  • Z-score analysis
  • Distribution profiling
• Detects:
  • Sudden spikes in null values
  • Dramatic shifts in value distributions
  • Upstream data problems

Delta Lake Quality Features

Continuous Monitoring

1. Track quality metrics over time
2. Set up alerts for quality degradation
3. Establish audit trails for compliance
4. Automate checks at every pipeline stage (ingestion → serving)

Data Quality Tools

Testing Pyramid

1. Unit tests — individual validation rules
2. Integration tests — pipeline validation
3. End-to-end tests — complete workflows
4. Production monitoring — continuous quality assurance

Key Takeaway: Each testing level provides different insights and catches different types of issues.

Advanced Techniques

• Data profiling — automatically discovers patterns and constraints
• Data observability — monitors health across entire data ecosystem
• Move from reactive fixes to proactive quality management

📊 Key Concepts Table

💻 Code Examples

Basic Data Validation

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("DataValidation") \
    .getOrCreate()

# Read data
df = spark.read \
    .format("json") \
    .load("/path/to/data")

# Define validation rules
# Rule 1: No null values in critical columns
# Parameter: isNotNull() — checks that column value is not null
df_not_null = df.filter(
    col("user_id").isNotNull() &
    col("email").isNotNull() &
    col("event_date").isNotNull()
)

# Rule 2: Data type validation
# Parameter: cast() — attempts to cast to target type, null if fails
df_type_valid = df.filter(
    col("amount").cast(DoubleType()).isNotNull() &
    col("event_date").cast(DateType()).isNotNull()
)

# Rule 3: Value range validation
# Parameter: between() — checks if value is within range
df_range_valid = df.filter(
    (col("amount") >= 0) &
    (col("amount") <= 1000000) &
    col("amount").between(0, 1000000)  # Alternative syntax
)

# Rule 4: Format validation using regex
# Parameter: rlike() — checks if string matches regex pattern
df_format_valid = df.filter(
    col("email").rlike(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$") &
    col("phone").rlike(r"^\+?[1-9]\d{1,14}$")  # E.164 format
)

# Combine all validations
df_valid = df.join(df_not_null.select("id"), "id", "inner") \
    .join(df_type_valid.select("id"), "id", "inner") \
    .join(df_range_valid.select("id"), "id", "inner") \
    .join(df_format_valid.select("id"), "id", "inner")

print(f"Original records: {df.count()}")
print(f"Valid records: {df_valid.count()}")
print(f"Validation rate: {df_valid.count() / df.count() * 100:.2f}%")

Statistical Anomaly Detection

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("AnomalyDetection") \
    .getOrCreate()

# Read data
df = spark.read \
    .format("delta") \
    .load("/path/to/data")

# Calculate statistics for anomaly detection
# Parameter: mean(), stddev() — aggregate functions for statistics
stats = df.select(
    mean("amount").alias("mean_amount"),
    stddev("amount").alias("stddev_amount"),
    count("amount").alias("count_amount"),
    min("amount").alias("min_amount"),
    max("amount").alias("max_amount")
).collect()[0]

# Apply Z-score anomaly detection
# Parameter: when().otherwise() — conditional column creation
df_with_anomalies = df.withColumn(
    "z_score",
    (col("amount") - lit(stats["mean_amount"])) / lit(stats["stddev_amount"])
).withColumn(
    "is_anomaly",
    when(abs(col("z_score")) > 3.0, True).otherwise(False)  # 3-sigma rule
).withColumn(
    "anomaly_type",
    when(col("z_score") > 3.0, "HIGH_VALUE")
    .when(col("z_score") < -3.0, "LOW_VALUE")
    .otherwise("NORMAL")
)

# Calculate anomaly statistics
anomaly_stats = df_with_anomalies.groupBy("is_anomaly").agg(
    count("*").alias("record_count"),
    avg("amount").alias("avg_amount"),
    stddev("amount").alias("stddev_amount")
)

print("Anomaly Statistics:")
anomaly_stats.show()

# Detect distribution anomalies using quantiles
# Parameter: approx_quantile() — approximate quantile calculation
quantiles = df.select(
    expr("percentile_approx(amount, 0.25)").alias("q1"),
    expr("percentile_approx(amount, 0.50)").alias("median"),
    expr("percentile_approx(amount, 0.75)").alias("q3")
).collect()[0]

iqr = quantiles["q3"] - quantiles["q1"]
lower_bound = quantiles["q1"] - 1.5 * iqr
upper_bound = quantiles["q3"] + 1.5 * iqr

# Mark IQR-based anomalies
df_iqr_anomalies = df.withColumn(
    "iqr_anomaly",
    when(
        (col("amount") < lower_bound) | (col("amount") > upper_bound),
        True
    ).otherwise(False)
)

print(f"IQR bounds: [{lower_bound:.2f}, {upper_bound:.2f}]")
print(f"IQR anomalies: {df_iqr_anomalies.filter(col('iqr_anomaly')).count()}")

Great Expectations Integration

from pyspark.sql import SparkSession
from great_expectations.dataset import SparkDFDataset

spark = SparkSession.builder \
    .appName("GreatExpectations") \
    .getOrCreate()

# Read data into Great Expectations
df = spark.read \
    .format("delta") \
    .load("/path/to/data")

# Wrap Spark DataFrame with Great Expectations
ge_df = SparkDFDataset(df)

# Define expectations
# Parameter: expect_column_values_to_not_be_null — null check
ge_df.expect_column_values_to_not_be_null("user_id")

# Parameter: expect_column_values_to_be_unique — uniqueness check
ge_df.expect_column_values_to_be_unique("email")

# Parameter: expect_column_values_to_be_in_set — value domain check
ge_df.expect_column_values_to_be_in_set(
    "status",
    ["active", "inactive", "pending", "suspended"]
)

# Parameter: expect_column_values_to_match_regex — format validation
ge_df.expect_column_values_to_match_regex(
    "email",
    r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
)

# Parameter: expect_column_values_to_be_between — range validation
ge_df.expect_column_values_to_be_between(
    "amount",
    min_value=0,
    max_value=1000000
)

# Parameter: expect_table_row_count_to_be_between — volume check
ge_df.expect_table_row_count_to_be_between(
    min_value=1000,
    max_value=1000000
)

# Run validation
results = ge_df.validate()

print(f"Validation success: {results.success}")
print(f"Successful expectations: {results.statistics['successful_expectations']}")
print(f"Unsuccessful expectations: {results.statistics['unsuccessful_expectations']}")

Deequ Validation Framework

from pyspark.sql import SparkSession
from com.amazon.deequ import VerificationSuite, VerificationResult
from com.amazon.deequ.checks import Check, CheckLevel
from com.amazon.deequ.constraints import Constraint

spark = SparkSession.builder \
    .appName("DeequValidation") \
    .getOrCreate()

# Read data
df = spark.read \
    .format("delta") \
    .load("/path/to/data")

# Define verification suite with checks
# Parameter: CheckLevel.Error — severity level for check failures
verification_suite = VerificationSuite() \
    .onData(df) \
    .addCheck(
        Check(spark, CheckLevel.Error, "Data Quality Check")
        # Null check
        .isComplete("user_id")
        .isComplete("email")
        .isComplete("event_date")
        
        # Uniqueness check
        .isUnique("email")
        
        # Range check
        .isContainedIn("amount", 0, 1000000)
        
        # Value set check
        .isContainedIn("status", ["active", "inactive", "pending"])
        
        # Compliance check (custom SQL)
        .satisfies(
            "event_date <= current_date()",
            "event_date_not_future",
            CheckLevel.Error,
            "Event date should not be in the future"
        )
        
        # Pattern check
        .matchesPattern(
            "email",
            r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
            "email_format"
        )
    )

# Run verification
result = verification_suite.run()

# Get results
if result.status == VerificationResult.SUCCESS:
    print("All checks passed!")
else:
    print("Some checks failed:")
    for check in result.checkResults:
        if check.status != "SUCCESS":
            print(f"  - {check.check}: {check.status}")

# Get metrics
metrics = result.metrics
print(f"\nMetrics:")
for metric_name, metric_value in metrics.items():
    print(f"  {metric_name}: {metric_value}")

📈 Performance Metrics

🏆 Best Practices

1. Define clear quality dimensions — Establish what quality means for your data
2. Implement validation at every stage — Ingestion, processing, and serving
3. Use statistical methods — Leverage Z-scores, distributions, and profiling
4. Automate quality checks — Embed validation in CI/CD pipelines
5. Monitor continuously — Track quality metrics over time
6. Set up alerting — Notify stakeholders of quality degradation
7. Maintain audit trails — Keep records of quality checks and results
8. Use appropriate tools — Great Expectations, Deequ, or custom solutions
9. Test with realistic data — Validate with production-like volumes
10. Iterate and improve — Refine rules based on feedback and issues

🔗 Related Topics

• 06-joins-optimization.mdx: Join quality and data consistency
• 07-partitioning-strategies.mdx: Partitioning for quality validation
• 14-merge-upsert.mdx: Quality during merge operations
• 16-schema-evolution.mdx: Schema quality and evolution

See Also

• 10-serialization-kryo.mdx: Serialization quality and consistency
• 11-structured-streaming.mdx: Streaming data quality validation
• Kafka Streams (kafka/03): Stream processing quality guarantees
• Data Engineering Streaming (data-engineering/022): Real-time data quality patterns