Handling the Chaos: Data Quality Strategies with PySpark Ingestion

dmatrixjsd — Sun, 16 Nov 2025 19:21:45 GMT

Tips and Techniques for Ingesting Large JSON files with PySpark

Introduction

Suppose you’ve ever struggled or grappled with consuming massive JSON files with PySpark. In that case, you are aware that insufficient data can always creep in and silently disrupt your ETL pipeline or pollute your data for downstream consumption.

In this blog post, I’ll walk you through three practical approaches to maintaining data quality during ingestion. We’ll cover how to drop records that don’t match your defined schema, validate that some numeric values are sensible (not just existent), and clean out any duplicate entries before they cause trouble downstream.

Figure 1. Overview of data quality strategies

These strategies seem simple yet effective techniques that can save you headaches when working with JSON files. While these three approaches are by no means exhaustive, other aspects of data quality, such as timeliness and consistency, will depend on data sources (real-time streaming, operational databases, etc.). Other external tools, such as Great Expectations, can also help mitigate these data validation issues.

For the past few weeks, I have been experimenting and cooperating with Claude Code, Cursor, and Goose, as my coding assistant: a generation and validation cycle, where I instruct, interact, validate, and reiterate with the AI coding agent for the best approach. Some of the code examples, test cases, and documentation are Cursor generated with my constant interactive approach of revision and validation, keeping AI “on a tight leash.”

Figure 2. Cycle of partial autonomy (source Andre Karpathy)

If you have not used the Cursor coding agent in this manner of cooperative partial autonomy, you ought to try it. 🤖 😎. You can explore the cooperative results of those data quality strategies in the GitHub repository.

Figure 3. PySpark files that implement the recommended strategy

Data Ingestion with Schema Enforcement

Consider schema enforcement. When working with large JSON files, one of the most critical steps is ensuring your data conforms to an expected schema before it pollutes your downstream processes. Schema enforcement in PySpark 4.0 DataFrame API gives you the power to catch malformed data early and handle it gracefully.

The key is being proactive rather than reactive. Instead of discovering schema mismatches after your ETL job has been running for hours, you can catch these issues upfront and decide how to handle them.

Example 1: Strict Schema Enforcement with Error Handling

Here’s how you can enforce a strict schema and capture any records that don’t comply. This strategy is a two-step process: FAILFAST for clean data; fallback to PERMISSIVE mode.

from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType from pyspark.sql.functions import col # Define schema once at module level for efficiency TRANSACTION_SCHEMA = StructType([ StructField("user_id", StringType(), nullable=False), StructField("transaction_amount", DoubleType(), nullable=False), StructField("transaction_date", TimestampType(), nullable=True), StructField("merchant_category", StringType(), nullable=True), StructField("customer_tier", StringType(), nullable=True) ]) def smart_json_load(spark, file_path): """ Smart JSON Loading: Try strict first, fallback to permissive. Best of both worlds: - Fast processing for clean data (FAILFAST) - Graceful recovery for messy data (PERMISSIVE) """ print(f"📋 Smart loading {file_path}...") # Step 1: Try FAILFAST for clean data try: print("🎯 Attempting FAILFAST mode...") df = spark.read.option("mode", "FAILFAST").schema(TRANSACTION_SCHEMA).json(file_path) count = df.count() # Force full evaluation to catch parsing errors - need to process all data try: df.collect() # This forces processing of ALL records, not just the first print(f"✅ FAILFAST succeeded: {count} clean records") return df, None except Exception as eval_error: # This catches parsing errors that occur during data access raise eval_error except Exception: print("⚠️ FAILFAST failed, switching to PERMISSIVE mode...") # Step 2: Fallback to PERMISSIVE schema_with_corrupt = schema.add(StructField("_corrupt_record", StringType(), True)) df = spark.read.option("mode", "PERMISSIVE").schema(schema_with_corrupt).json(file_path) # Cache to avoid Spark 4.0 corrupt record query restrictions df = df.cache() total_records = df.count() # Separate good from bad good_df = df.filter(col("_corrupt_record").isNull() & col("user_id").isNotNull()).drop("_corrupt_record") bad_df = df.filter(col("_corrupt_record").isNotNull() | col("user_id").isNull()) good_count = good_df.count() bad_count = bad_df.count() print(f"🔄 PERMISSIVE recovery: {good_count} good, {bad_count} bad records") return good_df, bad_df if __name__ == "__main__": spark = SparkSession.builder.appName("SmartJSONLoader").master("local[*]").getOrCreate() # Test with different data files test_files = [ "data/clean_transactions.json", "data/mixed_transactions.json", "data/invalid_transactions.json" ] for file_path in test_files: print(f"\n{'='*60}") print(f"🧪 TESTING: {file_path}") print(f"{'='*60}") good_data, bad_data = smart_json_load(spark, file_path) if bad_data is None: print(f"🎯 Perfect! Clean data processed with FAILFAST") else: print(f"🛡️ Recovered data with PERMISSIVE fallback") spark.stop()

Example 2: Permissive Mode with Corrupt Record Tracking

Sometimes you want to ingest what you can and track what fails. This example demonstrates PERMISSIVE mode in PySpark 4.0, which enables the ingestion of data that can be parsed while tracking problematic records for later analysis and investigation. Ideal for exploratory data analysis and gradual improvement of data quality.

from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType from pyspark.sql.functions import col # Define schema once at module level for efficiency PERMISSIVE_SCHEMA = StructType([ StructField("user_id", StringType(), nullable=True), StructField("transaction_amount", DoubleType(), nullable=True), StructField("transaction_date", TimestampType(), nullable=True), StructField("merchant_category", StringType(), nullable=True), StructField("customer_tier", StringType(), nullable=True), StructField("_corrupt_record", StringType(), nullable=True) # Captures bad records ]) def load_with_permissive_mode(spark, file_path): """ Example 2: Permissive Mode with Corrupt Record Tracking Uses PERMISSIVE mode - never fails, captures bad records in _corrupt_record column. Perfect for messy data where you want to save what you can. """ print(f"📋 Loading {file_path} with permissive mode...") # PERMISSIVE mode - never fails, captures corruption df = spark.read.option("mode", "PERMISSIVE").schema(PERMISSIVE_SCHEMA).json(file_path) total_records = df.count() # Separate good from bad records good_records = df.filter(col("_corrupt_record").isNull() & col("user_id").isNotNull()) bad_records = df.filter(col("_corrupt_record").isNotNull() | col("user_id").isNull()) good_count = good_records.count() bad_count = bad_records.count() success_rate = (good_count / total_records * 100) if total_records > 0 else 0 print(f"📊 Results: {good_count}/{total_records} good records ({success_rate:.1f}% success)") if good_count > 0: print("✅ Sample good records:") good_records.drop("_corrupt_record").show(2, truncate=False) if bad_count > 0: print("❌ Sample bad records:") bad_records.select("_corrupt_record").show(2, truncate=False) return good_records, bad_records if __name__ == "__main__": spark = SparkSession.builder.appName("PermissiveMode").master("local[*]").getOrCreate() # Test with different data files test_files = [ "data/clean_transactions.json", "data/mixed_transactions.json", "data/invalid_transactions.json" ] for file_path in test_files: print(f"\n{'='*60}") print(f"🧪 TESTING: {file_path}") print(f"{'='*60}") good_data, bad_data = load_with_permissive_mode(spark, file_path) print(f"✅ PERMISSIVE always succeeds - captured {good_data.count()} good, {bad_data.count()} bad records") spark.stop()

The beauty of this approach is that you can continue processing with the clean data while simultaneously capturing the problematic records for later analysis and review. This provides visibility into data quality issues without interrupting your entire pipeline. Code and test runs for the above examples are in this GitHub Repo.

Validating Data for Accuracy

The second approach is validating for data accuracy. Schema enforcement gets you part of the way there, but just because a field exists and has the correct data type doesn’t mean the data makes business sense. In this scenario, data validation ensures your transaction amounts fall within realistic ranges and your categorical values are from your approved business domains.

Think of this as your business logic validation layer. You’re not just checking that a transaction_amount field is a double; you’re checking that it’s a reasonable amount for a financial transaction. You’re not just verifying that merchant_category is a string; you’re ensuring it’s one of your supported merchant types like “grocery”, “restaurant”, or “retail” — not “currency”, which your payment system doesn’t handle.

Similarly, you’re not just confirming that customer_tier contains text data; you’re validating it matches your actual tier system (“bronze”, “silver”, “gold”, “platinum”) rather than made-up values like “diamond” that don’t exist in your business model. This validation layer catches data that’s technically valid but business-invalid:

❌ transaction_amount: -10.50 → Negative transactions don’t make sense
❌ transaction_amount: 15000.00 → Exceeds your transaction limits
❌ merchant_category: “currency” → Not in your supported merchant types
❌ customer_tier: “diamond” → Not in your tier system

Schema validation ensures data structure integrity; business logic validation ensures that the data makes sense within your business context.

Example 1: Range and Business Logic Validation

from pyspark.sql import SparkSession from pyspark.sql.functions import col, when from pyspark.sql.types import StructType, StructField, StringType, DoubleType def validate_transaction_business_rules(df): """ Business Logic Validation - ensuring data makes business sense and it goes beyond schema validation to check ranges, categories, and business constraints. """ # Define business rules VALID_MERCHANTS = ["grocery", "restaurant", "gas_station", "retail", "entertainment", "other"] VALID_TIERS = ["bronze", "silver", "gold", "platinum"] MIN_AMOUNT = 0.01 MAX_AMOUNT = 10000.0 # Apply validation rules validated_df = df.withColumn( "validation_status", when(col("transaction_amount") < MIN_AMOUNT, "negative_or_zero_amount") .when(col("transaction_amount") > MAX_AMOUNT, "excessive_amount") .when(~col("merchant_category").isin(VALID_MERCHANTS), "invalid_merchant_category") .when(~col("customer_tier").isin(VALID_TIERS), "invalid_customer_tier") .otherwise("valid") ) # Separate valid from invalid data valid_df = validated_df.filter(col("validation_status") == "valid").drop("validation_status") invalid_df = validated_df.filter(col("validation_status") != "valid") # Show validation summary total = validated_df.count() valid_count = valid_df.count() invalid_count = invalid_df.count() success_rate = (valid_count / total * 100) if total > 0 else 0 print(f"📊 Business Logic Validation: {valid_count}/{total} valid ({success_rate:.1f}%)") if invalid_count > 0: print("🔍 Validation issues:") validated_df.groupBy("validation_status").count().show() return valid_df, invalid_df if __name__ == "__main__": # Test the validation function spark = SparkSession.builder.appName("BusinessLogicValidation").master("local[*]").getOrCreate() # Create test data with various business rule violations test_data = [ # Valid records ("user_001", 45.99, "grocery", "gold"), ("user_002", 125.50, "restaurant", "silver"), ("user_003", 89.99, "retail", "bronze"), # Invalid records ("user_004", -10.50, "grocery", "gold"), # Negative amount ("user_005", 15000.00, "restaurant", "silver"), # Excessive amount ("user_006", 50.00, "currency", "gold"), # Invalid merchant ("user_007", 75.25, "grocery", "diamond"), # Invalid tier ("user_008", 0.00, "gas_station", "platinum"), # Zero amount ] schema = StructType([ StructField("user_id", StringType(), False), StructField("transaction_amount", DoubleType(), False), StructField("merchant_category", StringType(), False), StructField("customer_tier", StringType(), False) ]) test_df = spark.createDataFrame(test_data, schema) print("🧪 Testing Business Logic Validation") print("=" * 50) # Apply validation valid_data, invalid_data = validate_transaction_business_rules(test_df) # Verify results if valid_data.count() == 3 and invalid_data.count() == 5: print("✅ Test PASSED: 3 valid, 5 invalid records as expected") else: print(f"❌ Test FAILED: Expected 3 valid, 5 invalid. Got {valid_data.count()} valid, {invalid_data.count()} invalid") spark.stop()

The code example to validate business logic and data validation, and its relevant test run for the above example, is in this GitHub Repo.

In addition to schema and business logic validation and data consistency, another common source of data chaos is data duplication.

Dropping Data Duplicates

The final approach is to consider duplicates. Duplicate records can skew your analytics and reporting. They can inflate metrics, create incorrect aggregations, and lead to poor business decisions. PySpark DataFrame API provides powerful deduplication capabilities that go beyond simple row-by-row comparisons.

The key insight here is that duplicates aren’t always exact row matches. Sometimes, you need to define business logic around what constitutes a duplicate — perhaps it’s the same transaction ID, or the same customer making the same purchase within a specific time window.

Let’s examine the code for both scenarios: simple exact matches and business logic that dictates duplicates.

Example 1: Simple Deduplication with Performance Optimization

from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType from pyspark.sql.functions import col, row_number, desc, when, lag, unix_timestamp, abs as spark_abs from pyspark.sql.window import Window from datetime import datetime def create_sample_data(spark): """Create sample data with various duplicate patterns""" data = [ # Exact duplicates ("TXN001", "USER123", 99.99, datetime(2024, 1, 15, 10, 30, 0), "RETAIL", "GOLD"), ("TXN001", "USER123", 99.99, datetime(2024, 1, 15, 10, 30, 0), "RETAIL", "GOLD"), # Same transaction_id, different details (system error) ("TXN002", "USER456", 150.00, datetime(2024, 1, 15, 11, 0, 0), "GROCERY", "SILVER"), ("TXN002", "USER789", 200.00, datetime(2024, 1, 15, 11, 5, 0), "RETAIL", "BRONZE"), # Same user + amount within 30 seconds (double-click) ("TXN003", "USER123", 49.99, datetime(2024, 1, 15, 12, 0, 0), "ONLINE", "GOLD"), ("TXN004", "USER123", 49.99, datetime(2024, 1, 15, 12, 0, 30), "ONLINE", "GOLD"), # Unique records ("TXN005", "USER456", 75.50, datetime(2024, 1, 15, 13, 0, 0), "RESTAURANT", "SILVER"), ("TXN006", "USER789", 120.00, datetime(2024, 1, 15, 14, 0, 0), "GAS", "BRONZE"), ] schema = StructType([ StructField("transaction_id", StringType(), False), StructField("user_id", StringType(), False), StructField("transaction_amount", DoubleType(), False), StructField("transaction_date", TimestampType(), False), StructField("merchant_category", StringType(), True), StructField("customer_tier", StringType(), True) ]) return spark.createDataFrame(data, schema) def exact_deduplication(df): """Remove exact duplicate rows - fastest method""" return df.distinct() def business_logic_deduplication(df): """Remove duplicates based on business rules""" # Rule 1: Same transaction_id = duplicate (keep first occurrence) window_txn = Window.partitionBy("transaction_id").orderBy("transaction_date") df = df.withColumn("txn_rank", row_number().over(window_txn)) \ .filter(col("txn_rank") == 1) \ .drop("txn_rank") # Rule 2: Same user + amount within 60 seconds = double-click (keep first) window_user = Window.partitionBy("user_id", "transaction_amount").orderBy("transaction_date") df = df.withColumn("prev_time", lag("transaction_date").over(window_user)) \ .withColumn("time_diff", when(col("prev_time").isNull(), 999999) .otherwise(spark_abs(unix_timestamp("transaction_date") - unix_timestamp("prev_time")))) \ .filter(col("time_diff") > 60) \ .drop("prev_time", "time_diff") return df def priority_deduplication(df): """Keep the best record when duplicates exist""" # Priority: GOLD > SILVER > BRONZE df_with_priority = df.withColumn("priority", when(col("customer_tier") == "GOLD", 3) .when(col("customer_tier") == "SILVER", 2) .when(col("customer_tier") == "BRONZE", 1) .otherwise(0)) window_priority = Window.partitionBy("transaction_id") \ .orderBy(desc("priority"), desc("transaction_date")) return df_with_priority.withColumn("rank", row_number().over(window_priority)) \ .filter(col("rank") == 1) \ .drop("priority", "rank") if __name__ == "__main__": # Initialize Spark with performance optimizations spark = SparkSession.builder \ .appName("DeduplicationExample") \ .master("local[*]") \ .config("spark.sql.adaptive.enabled", "true") \ .getOrCreate() spark.sparkContext.setLogLevel("ERROR") # Reduce verbose output # Create sample data df = create_sample_data(spark) print(f"Original records: {df.count()}") # Method 1: Exact deduplication (fastest for exact matches) exact_result = exact_deduplication(df) print(f"After exact deduplication: {exact_result.count()}") # Method 2: Business logic deduplication business_result = business_logic_deduplication(df) print(f"After business logic deduplication: {business_result.count()}") # Method 3: Priority-based deduplication priority_result = priority_deduplication(df) print(f"After priority deduplication: {priority_result.count()}") # Show final results print("\nFinal deduplicated data:") business_result.show(truncate=False) spark.stop()

Example 2: Advanced Deduplication with Custom Logic

On occasions, you need more sophisticated deduplication logic — perhaps keeping the record with the highest value, the most recent timestamp, using multi-tier priority ranking, or aggregating instead of removing entries. Some of this custom logic will be dependent on your business use case.

from pyspark.sql import SparkSession from pyspark.sql.functions import col, row_number, desc, when, max as spark_max, sum as spark_sum, count from pyspark.sql.window import Window from datetime import datetime def create_sample_data(spark): """Sample customer data with duplicates""" data = [ ("CUST001", "John Smith", 50.00, datetime(2024, 1, 15, 10, 0, 0), "Basic", "ACTIVE"), ("CUST001", "John Smith", 150.00, datetime(2024, 1, 15, 11, 0, 0), "Premium", "ACTIVE"), ("CUST002", "Jane Doe", 100.00, datetime(2024, 1, 15, 9, 0, 0), "Standard", "SUSPENDED"), ("CUST002", "Jane Doe", 100.00, datetime(2024, 1, 15, 11, 0, 0), "Standard", "ACTIVE"), ("CUST003", "Bob Wilson", 220.00, datetime(2024, 1, 15, 8, 0, 0), "Basic", "ACTIVE"), ("CUST003", "Bob Wilson", 180.00, datetime(2024, 1, 15, 10, 0, 0), "Premium", "ACTIVE"), ] columns = ["customer_id", "name", "amount", "timestamp", "tier", "status"] return spark.createDataFrame(data, columns) def keep_highest_value(df): """Keep record with highest amount per customer""" window = Window.partitionBy("customer_id").orderBy(desc("amount")) return df.withColumn("rank", row_number().over(window)) \ .filter(col("rank") == 1).drop("rank") def keep_most_recent(df): """Keep most recent record per customer""" window = Window.partitionBy("customer_id").orderBy(desc("timestamp")) return df.withColumn("rank", row_number().over(window)) \ .filter(col("rank") == 1).drop("rank") def multi_criteria_best(df): """Advanced: Keep best record using business priority""" df_priority = df.withColumn("tier_score", when(col("tier") == "Premium", 3) .when(col("tier") == "Standard", 2) .otherwise(1)) \ .withColumn("status_score", when(col("status") == "ACTIVE", 1).otherwise(0)) window = Window.partitionBy("customer_id") \ .orderBy(desc("tier_score"), desc("status_score"), desc("amount"), desc("timestamp")) return df_priority.withColumn("rank", row_number().over(window)) \ .filter(col("rank") == 1) \ .drop("tier_score", "status_score", "rank") def aggregate_duplicates(df): """Alternative: Aggregate instead of removing duplicates""" return df.groupBy("customer_id", "name") \ .agg(spark_max("amount").alias("max_amount"), spark_sum("amount").alias("total_amount"), count("*").alias("record_count"), spark_max("timestamp").alias("latest_update")) if __name__ == "__main__": spark = SparkSession.builder.appName("AdvancedDedup").master("local[*]").getOrCreate() spark.sparkContext.setLogLevel("ERROR") df = create_sample_data(spark) print(f"Original: {df.count()} records") # Method 1: Keep highest value highest = keep_highest_value(df) print(f"Highest value: {highest.count()} records") highest.select("customer_id", "name", "amount", "tier").show() # Method 2: Keep most recent recent = keep_most_recent(df) print(f"Most recent: {recent.count()} records") recent.select("customer_id", "name", "timestamp", "status").show() # Method 3: Multi-criteria priority best = multi_criteria_best(df) print(f"Multi-criteria best: {best.count()} records") best.show() # Method 4: Aggregate instead of remove agg = aggregate_duplicates(df) print(f"Aggregated: {agg.count()} records") agg.show() spark.stop()

For a comprehensive set of deduplication strategies with custom logic, refer to the GitHub Repository. These approaches give you control over which records to keep when duplicates are found, rather than just arbitrarily keeping the first occurrence. For a more advanced version of deduplication, check the code repository.

Conclusion

To recap, data quality issues in large JSON ingestion pipelines are inevitable, but they don’t have to be catastrophic. By implementing these three strategies — schema enforcement, data validation, and custom deduplication — you can identify and address problems early, handling them with greater ease. By no means are these exhaustive, yet they are sufficient and commonly used.

The key is to be proactive rather than reactive. Set up your quality checks at ingestion time, not after your data has already propagated through your entire data lake. These PySpark techniques provide you with the tools to build robust data quality pipelines that can handle the complexities of real-world data.

Remember that data quality is an ongoing process, not a one-time fix. Monitor your validation metrics over time and continuously refine your validation rules as your understanding of the data evolves.

Your downstream consumers will thank you for the clean, reliable data, not least for decreasing the frequency of your 3 AM pager duty calls.

Resources and References

Apache Spark™ 4.0 Documentation Overview — The main documentation hub for Apache Spark™ 4.0
https://spark.apache.org/docs/latest/
PySpark 4.0 API Documentation — Official Python API documentation for Apache Spark™ 4.0
https://spark.apache.org/docs/latest/api/python/index.html
Delta Lake Schema Enforcement and Evolution — Comprehensive guide on schema enforcement mechanisms in Delta Lake
https://www.databricks.com/blog/2019/09/24/diving-into-delta-lake-schema-enforcement-evolution.html
Databricks Schema Enforcement Documentation — Best practices for data quality through schema enforcement
https://docs.databricks.com/aws/en/tables/schema-enforcement
Introducing Apache Spark™ 4.0 — Overview of key features and improvements in Spark 4.0
https://www.databricks.com/blog/introducing-apache-spark-40
Great Expectations — Data validation and documentation tool for data quality management
https://greatexpectations.io/

topic Handling the Chaos: Data Quality Strategies with PySpark Ingestion in Community Articles