3.3.4. Handling Data Quality Issues

💡 First Principle: Real-world data contains duplicates, nulls, and late arrivals. Robust data engineering handles these systematically rather than hoping they don't exist.

Handling Duplicates

# PySpark: Remove duplicates
df_deduped = df.dropDuplicates(["OrderID"])  # subset list; with no list every column must match, and distinct() takes no subset

# Keep most recent duplicate
from pyspark.sql.window import Window
window = Window.partitionBy("OrderID").orderBy(col("UpdatedAt").desc())
df_latest = df.withColumn("row_num", row_number().over(window)) \
              .filter(col("row_num") == 1) \
              .drop("row_num")

⚠️ Silent duplicates from a broken upsert: if a MERGE/upsert match condition stops matching (for example, the source system changes its key format), every incoming row takes the INSERT branch. New duplicate rows pile up while the original rows keep their old values. When duplicates appear with no code change, check the MERGE ON condition against the current source data.

Handling Missing Data

# Fill nulls with default values (df.na.fill() and df.fillna() are the same call;
# F.coalesce(col, lit(default)) does it per column)
df_filled = df.na.fill({
    "Quantity": 0,
    "Discount": 0.0,
    "ShipDate": "1900-01-01"
})

# Drop rows with nulls in critical columns
df_complete = df.na.drop(subset=["CustomerID", "ProductID"])

Handling Late-Arriving Data

• Challenge: Data arrives after the batch window closes
• Solutions:
  • Watermarks: Track latest processed timestamp, and re-read a lookback window (for example, last watermark minus 2 days). A strict > last max timestamp filter skips rows that arrive late with older timestamps; MERGE or dedupe the overlap
  • Reprocessing windows: Reprocess recent data periodically
  • Delta Lake merge: Update existing records with late data

Validation and lineage: Validate and standardize data types (cast, trim, consistent date formats) for every source, including "trusted" ones, and keep typed columns rather than casting everything to string; failing rows go to quarantine, not deletion. For audit lineage, stamp each row with the source name, pipeline run ID and load timestamp, passed in as pipeline parameters.

Validation Patterns (PySpark)

Reject, don't drop (the dead-letter pattern, batch or streaming): send records that fail parsing or validation to a quarantine (reject, dead-letter) table or error folder with the failure reason, keep loading the good rows, and log the reject count, so the failures can be fixed and reprocessed later. Silently dropping rows hides data loss and removes the evidence needed for root-cause analysis and for reconciling with the source.

Data Quality Dimensions

A dataset can be complete (no missing rows or nulls) and still inaccurate (wrong values).