Introduction
Bad data silently corrupts pipelines. By the time you discover a schema mismatch or missing value in production, it's already propagated downstream, failed dashboards, and broken decisions. Data quality checks: a practical Spark & Data Eng guide covers the defensive patterns we've used across trading platforms, e-commerce warehouses, and ride-hailing systems to catch problems early, quarantine bad records, and keep pipelines reliable.
Data quality isn't a luxury. It's the foundation of trust in every system that touches data. In this post, we'll walk through schema validation, value-level assertions, tooling choices, and quarantine strategies that actually work in production.
Schema checks — required columns and types
Schema mismatches are the most common upstream surprise. A CSV arrives with columns in the wrong order, a JSON feed adds an unexpected field, or a third-party database changes a column type without notice. By the time your pipeline crashes, you've lost time and trust.
Start with explicit schema definition. Rather than inferring schema from data, declare it upfront in your Spark code. This catches misalignment immediately.
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, TimestampType
# Define explicit schema for order data
orders_schema = StructType([
StructField("order_id", StringType(), nullable=False),
StructField("customer_id", StringType(), nullable=False),
StructField("order_date", TimestampType(), nullable=False),
StructField("amount", DoubleType(), nullable=False),
StructField("status", StringType(), nullable=False),
])
# Read with schema enforcement
orders_df = spark.read \
.schema(orders_schema) \
.option("mode", "failFast") \
.csv("s3://bucket/orders.csv")
The failFast mode stops immediately on the first schema violation. If you need to be more lenient during development or testing, use permissive mode and validate separately, but always tighten mode in production.
Next, validate that required columns exist and no unexpected columns have appeared:
def validate_schema(df, required_columns, expected_columns=None):
"""Check for required and unexpected columns."""
actual_columns = set(df.columns)
required_set = set(required_columns)
# Check required columns present
missing = required_set - actual_columns
if missing:
raise ValueError(f"Missing required columns: {missing}")
# Check for unexpected columns (schema drift)
if expected_columns:
expected_set = set(expected_columns)
unexpected = actual_columns - expected_set
if unexpected:
raise ValueError(f"Unexpected columns detected (schema drift): {unexpected}")
return True
# Usage
validate_schema(
orders_df,
required_columns=["order_id", "customer_id", "order_date", "amount", "status"],
expected_columns=["order_id", "customer_id", "order_date", "amount", "status", "notes"]
)
This pattern catches both missing required columns and schema drift—when upstream adds fields you didn't expect. In mature systems, schema drift often signals an upstream change that needs communication.
Value checks — nulls, ranges, and uniqueness
Even with correct schemas, individual values can be invalid. Nulls in non-nullable fields, negative amounts, future dates, or duplicate IDs that should be unique. These slip through at scale and corrupt fact tables, revenue calculations, and identity models.
Start by auditing null counts per column:
def check_nulls(df, columns, max_null_percent=0.01):
"""
Check null counts per column.
Raise if any column exceeds max_null_percent (default 1%).
"""
total_rows = df.count()
issues = []
for col in columns:
null_count = df.filter(df[col].isNull()).count()
null_pct = null_count / total_rows
if null_pct > max_null_percent:
issues.append(
f"{col}: {null_count} nulls ({null_pct*100:.2f}%) exceeds {max_null_percent*100:.1f}%"
)
if issues:
raise ValueError("Null checks failed:\n" + "\n".join(issues))
return True
# Usage
check_nulls(
orders_df,
columns=["order_id", "customer_id", "amount"],
max_null_percent=0.005
)
For value ranges, enforce business logic constraints:
def check_value_ranges(df, checks):
"""
checks: dict of column -> (min_val, max_val) or list of conditions
Example: {"amount": (0, 1_000_000), "discount_pct": (0, 100)}
"""
issues = []
for col, (min_val, max_val) in checks.items():
invalid = df.filter((df[col] < min_val) | (df[col] > max_val))
invalid_count = invalid.count()
if invalid_count > 0:
issues.append(f"{col}: {invalid_count} records outside [{min_val}, {max_val}]")
if issues:
raise ValueError("Range checks failed:\n" + "\n".join(issues))
return True
# Usage
check_value_ranges(
orders_df,
checks={
"amount": (0.01, 1_000_000),
"discount_pct": (0, 100),
}
)
For uniqueness, identify duplicates that shouldn't exist:
def check_uniqueness(df, columns, allow_duplicates=False):
"""Check that specified columns are unique (or count duplicates)."""
if allow_duplicates:
# Just report
grouped = df.groupBy(columns).count()
dupes = grouped.filter(grouped["count"] > 1)
dupe_count = dupes.count()
print(f"Found {dupe_count} duplicate groups in {columns}")
return dupes
else:
# Strict: no duplicates allowed
grouped = df.groupBy(columns).count()
dupes = grouped.filter(grouped["count"] > 1)
dupe_count = dupes.count()
if dupe_count > 0:
raise ValueError(f"Uniqueness check failed: {dupe_count} duplicate combinations in {columns}")
return True
# Usage: order_id must be globally unique
check_uniqueness(orders_df, columns=["order_id"], allow_duplicates=False)
Tools — Great Expectations, Deequ, or dbt tests
Manual checks work for small teams, but they're labor-intensive and easy to forget. Professional data stacks use dedicated tools.
Great Expectations
Great Expectations gives you a declarative, reusable framework for data validation. Define expectations once, run them everywhere:
from great_expectations.dataset.sparkdf_dataset import SparkDFDataset
# Wrap Spark DataFrame
ge_df = SparkDFDataset(orders_df)
# Define expectations
expectations = [
ge_df.expect_column_to_exist("order_id"),
ge_df.expect_column_values_to_not_be_null("customer_id"),
ge_df.expect_column_values_to_be_in_set("status", ["pending", "confirmed", "shipped", "delivered"]),
ge_df.expect_column_values_to_be_between("amount", min_value=0, max_value=1_000_000),
]
# Run validation
results = ge_df.validate(expectations, result_format="SUMMARY")
print(results)
if not results["success"]:
raise ValueError(f"Data quality check failed: {results}")
AWS Deequ
For teams deep in the AWS ecosystem, Deequ (from Amazon) is a lightweight PySpark library that scales validation across large datasets:
from deequ.checks import Check, CheckLevel
from deequ.verification import VerificationSuite, VerificationResult
# Define checks
checks = [
Check(CheckLevel.Error, "Order Schema") \
.hasSize(lambda x: x > 0) \
.isComplete("order_id") \
.isUnique("order_id"),
Check(CheckLevel.Warning, "Order Values") \
.isInRange("amount", 0, 1_000_000) \
.isContainedIn("status", ["pending", "confirmed", "shipped", "delivered"]),
]
# Run verification
verification_result = VerificationSuite(spark) \
.onData(orders_df) \
.addChecks(checks) \
.run()
# Handle results
if verification_result.status.name != "Success":
print(verification_result.checkResults)
dbt tests
If you're already on dbt, leverage its testing framework. It's lighter weight and integrates naturally with your transformation DAG:
# models/staging/stg_orders.yml
version: 2
models:
- name: stg_orders
description: Raw orders staged for transformation
columns:
- name: order_id
description: Unique order identifier
tests:
- unique
- not_null
- name: customer_id
tests:
- not_null
- name: amount
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_of_type:
column_type: numeric
- dbt_expectations.expect_column_values_to_be_between:
min_value: 0
max_value: 1000000
- name: status
tests:
- not_null
- accepted_values:
values: ['pending', 'confirmed', 'shipped', 'delivered']
Run tests as part of your CI/CD: dbt test will fail the pipeline if any assertion breaks.
Quarantine — route bad records to a separate table
When data fails validation, you have two choices: fail the entire pipeline or quarantine bad records for manual inspection. Production systems need both—fail fast for critical issues, but quarantine for inspection and recovery.
Implement a quarantine pattern:
def validate_and_split(df, validation_func):
"""
Run validation_func on df.
Return (good_df, bad_df, validation_report)
"""
bad_records = []
good_records = []
# Validate row-by-row (for demonstration; optimize with UDFs for scale)
for row in df.collect():
try:
validation_func(row)
good_records.append(row)
except Exception as e:
bad_records.append((row, str(e)))
good_df = spark.createDataFrame(good_records, schema=df.schema) if good_records else None
# Create quarantine table with error messages
if bad_records:
quarantine_schema = StructType(
list(df.schema.fields) + [
StructField("error_message", StringType()),
StructField("quarantine_timestamp", TimestampType()),
]
)
quarantine_rows = [
(*row, error_msg, datetime.now())
for row, error_msg in bad_records
]
bad_df = spark.createDataFrame(quarantine_rows, schema=quarantine_schema)
else:
bad_df = None
return good_df, bad_df, len(bad_records)
# Usage
def validate_order_row(row):
"""Custom validation for single row."""
if row.amount < 0 or row.amount > 1_000_000:
raise ValueError(f"Amount {row.amount} out of range")
if row.status not in ["pending", "confirmed", "shipped", "delivered"]:
raise ValueError(f"Invalid status: {row.status}")
good_orders, bad_orders, bad_count = validate_and_split(orders_df, validate_order_row)
# Write good records to production table
good_orders.write.mode("overwrite").parquet("s3://bucket/orders/good/")
# Write bad records to quarantine table for investigation
if bad_orders:
bad_orders.write.mode("append").parquet("s3://bucket/orders/quarantine/")
print(f"Quarantined {bad_count} bad records for review")
For higher scale, use a UDF-based approach:
from pyspark.sql.functions import udf, col, when
from pyspark.sql.types import BooleanType, StringType
def validate_order_udf(order_id, amount, status):
"""Validate single order, return (is_valid, error_message)."""
if not order_id:
return (False, "Missing order_id")
if amount < 0 or amount > 1_000_000:
return (False, f"Amount {amount} out of range")
if status not in ["pending", "confirmed", "shipped", "delivered"]:
return (False, f"Invalid status: {status}")
return (True, None)
# Apply validation UDF
from pyspark.sql.functions import col, struct
validation_udf = udf(validate_order_udf, "struct")
validated_df = orders_df.withColumn(
"validation",
validation_udf(col("order_id"), col("amount"), col("status"))
)
good_df = validated_df.filter(col("validation.is_valid") == True).drop("validation")
bad_df = validated_df.filter(col("validation.is_valid") == False) \
.withColumn("error_message", col("validation.error_message")) \
.drop("validation")
good_df.write.mode("overwrite").parquet("s3://bucket/orders/good/")
bad_df.write.mode("append").parquet("s3://bucket/orders/quarantine/")
Store quarantine records with timestamps and error details. Set up a monitoring alert when quarantine volume spikes—it signals upstream issues.
Common mistakes
- Skipping schema validation in development. "It works on my machine" leads to surprise failures in production. Always validate schema, even if you're reading from a trusted source.
- Setting thresholds too tight. If your null check fails at 0.1% nulls but your data naturally has 0.5%, you'll get alert fatigue. Baseline your data first, then set thresholds that catch real anomalies.
- Failing the entire pipeline on one bad record. For high-volume transactional systems, quarantine bad records and process good ones. Reserve hard failures for schema breaks or systematic failures.
- Forgetting to validate after transformation. Data quality checks belong at data entry (source validation) and after each major transformation. A bad join or aggregation can corrupt clean input.
- Not monitoring quarantine tables. A quarantine table that grows silently is worthless. Wire up metrics and alerts so you know when bad data is accumulating.
Key takeaways
- Define schemas explicitly and validate on every read. Use
failFastmode in production to catch schema mismatches immediately. - Implement row-level checks for nulls, ranges, and uniqueness. Use UDFs at scale to avoid expensive collect() calls.
- Choose a validation tool—Great Expectations for flexibility, Deequ for AWS scale, or dbt tests for transformation-native workflows.
- Quarantine bad records instead of failing the entire pipeline. Route them to a separate table with timestamps and error messages for investigation.
- Monitor quarantine tables and alert on volume spikes. Silent data quality degradation is worse than pipeline failures.
- Validate after every transformation, not just at the source. Bad logic upstream corrupts everything downstream.
Comments
Loading comments…