Late Arriving Dimensions: A Production PySpark Guide (TB-Scale, S3 → Snowflake)

Late Arriving Dimensions is one of the trickiest real-world data warehousing problems. It occurs when fact events arrive before their corresponding dimension records exist in the warehouse. For example, a sales transaction lands in your pipeline today referencing product_id = P9087, but the product master feed containing P9087 hasn't been loaded yet — it arrives 6 hours later due to a delayed upstream extract.

This guide walks through a production-grade, TB-scale implementation using pure PySpark (no Delta/Hudi/Iceberg), reading from S3 Parquet and writing to Snowflake.

What Causes Late Arriving Dimensions?

In enterprise data platforms, fact streams and dimension feeds are typically produced by different source systems with different SLAs:

When the fact references a dimension key that doesn't yet exist in the dimension table, the foreign key lookup fails, and the fact record is left with a broken or missing surrogate key.

Why Can't We Just "Wait" for the Dimension?

At TB-scale, holding back billions of fact rows until every dimension is confirmed is not practical:

1. SLA Pressure: Downstream reports and dashboards depend on fact data being available by a strict time (e.g., 6:00 AM). You can't delay terabytes of transactions because one product feed is late.
2. Unpredictable Delays: Some dimension feeds are late by hours, some by days. You'd need unbounded buffering.
3. Cascading Failures: Blocking the entire pipeline for one dimension means no data flows for any consumer — analytics, ML, finance, all stalled.

The Solution: Placeholder + Reconciliation Pattern

The industry-standard approach has two phases:

Phase 1 — Initial Load (Fact Arrives First)

1. Attempt to look up the dimension surrogate key (SK) for each fact record.
2. If the dimension key is found → assign the correct SK.
3. If the dimension key is NOT found → assign a placeholder SK (typically -1 or a dedicated "Unknown" row in the dimension table) and flag the record for later reconciliation.

Phase 2 — Reconciliation (Dimension Arrives Late)

1. After the late dimension feed lands, a scheduled reconciliation job scans the fact table for records still carrying the placeholder SK.
2. The job re-attempts the dimension lookup using the now-available dimension data.
3. Matched records are updated in-place in Snowflake with the correct SK.
4. Unmatched records remain flagged for the next reconciliation cycle.

Architecture Flow

The following diagram illustrates the complete Late Arriving Dimensions lifecycle — fact records landing with placeholder keys, late dimension arrival, PySpark reconciliation, and corrected writes to Snowflake:

Step-by-Step Implementation

Environment Assumptions

• Fact table size: ~2.5 TB daily, partitioned by order_date in S3 as Parquet
• Dimension table: ~50 GB customer dimension, stored in Snowflake (DW.DIM_CUSTOMER)
• Fact target: Snowflake (DW.FACT_ORDERS)
• Source landing zone: s3://datalake-raw/facts/orders/ and s3://datalake-raw/dimensions/customers/
• PySpark cluster: EMR with 20 executors × 5 cores × 16GB memory

Step 1: Set Up Spark Session with Snowflake Connector

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *
from pyspark.sql.window import Window

spark = SparkSession.builder \
    .appName("Late-Arriving-Dimensions-Reconciliation") \
    .config("spark.jars", "/opt/spark/jars/spark-snowflake_2.12-2.11.0.jar,/opt/spark/jars/snowflake-jdbc-3.13.22.jar") \
    .config("spark.sql.shuffle.partitions", 800) \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .getOrCreate()

# Snowflake connection options (use secrets manager in production)
SNOWFLAKE_OPTIONS = {
    "sfURL": "myorg-myaccount.snowflakecomputing.com",
    "sfUser": "ETL_SERVICE_ACCOUNT",
    "sfPassword": "********",        # Use AWS Secrets Manager in real environments
    "sfDatabase": "DW",
    "sfSchema": "PUBLIC",
    "sfWarehouse": "ETL_WH_LARGE",   # XL warehouse for TB-scale writes
    "sfRole": "ETL_ROLE",
}

Step 2: Read Today's Fact Data from S3

The fact stream lands as partitioned Parquet files in S3. At 2.5 TB, this is a significant read — we use predicate pushdown on the partition column to limit scope.

# -------------------------------------------------------
# Read today's fact events from S3 Parquet landing zone
# Partitioned by order date for efficient filtering
# -------------------------------------------------------
FACT_S3_PATH = "s3://datalake-raw/facts/orders/"
TODAY = "2026-05-31"

facts_df = spark.read.parquet(FACT_S3_PATH) \
    .filter(F.col("order_date") == TODAY)

print(f"📥 Facts loaded: {facts_df.count():,} records for {TODAY}")
facts_df.printSchema()
facts_df.show(5, truncate=False)

Sample Fact Data (as received from source):

+----------+----------+-----------+--------+--------+-------------------+
|order_id  |order_date|customer_id|product |amount  |event_timestamp    |
+----------+----------+-----------+--------+--------+-------------------+
|ORD-90001 |2026-05-31|C101       |Laptop  |85000.00|2026-05-31 09:15:22|
|ORD-90002 |2026-05-31|C102       |Phone   |45000.00|2026-05-31 09:20:18|
|ORD-90003 |2026-05-31|C777       |Tablet  |32000.00|2026-05-31 09:25:40|  ← C777 NOT in dimension yet!
|ORD-90004 |2026-05-31|C888       |Monitor |28000.00|2026-05-31 09:30:55|  ← C888 NOT in dimension yet!
|ORD-90005 |2026-05-31|C103       |Keyboard| 3500.00|2026-05-31 09:45:10|
+----------+----------+-----------+--------+--------+-------------------+

Notice: C777 and C888 are new customers whose dimension records haven't arrived yet.

Step 3: Read the Current Customer Dimension from Snowflake

# -------------------------------------------------------
# Read the current customer dimension table from Snowflake
# This is ~50 GB — Spark will parallelize the JDBC read
# -------------------------------------------------------
dim_customer_df = spark.read \
    .format("snowflake") \
    .options(**SNOWFLAKE_OPTIONS) \
    .option("dbtable", "DIM_CUSTOMER") \
    .option("autopushdown", "on") \
    .load() \
    .filter(F.col("is_current") == True) \
    .select(
        F.col("surrogate_key").alias("customer_sk"),
        F.col("customer_id"),
        F.col("customer_name"),
        F.col("city"),
        F.col("tier")
    )

# Cache the dimension since we'll use it multiple times
dim_customer_df.cache()
print(f"📦 Dimension loaded: {dim_customer_df.count():,} active customer records")

Step 4: Phase 1 — Initial Fact Load with Placeholder Assignment

This is the core logic. We attempt a left join between facts and the dimension. Where the join succeeds, we get the real surrogate key. Where it fails (dimension not yet available), we assign -1 as a placeholder and flag the record.

# -------------------------------------------------------
# PHASE 1: Attempt dimension lookup
# Left join facts with dimension on natural key (customer id)
# -------------------------------------------------------
PLACEHOLDER_SK = -1

enriched_df = facts_df.alias("f").join(
    dim_customer_df.alias("d"),
    on=F.col("f.customer_id") == F.col("d.customer_id"),
    how="left"
)

# Assign the surrogate key or placeholder
# Also flag whether the dimension was found or not
facts_with_sk = enriched_df.select(
    F.col("f.order_id"),
    F.col("f.order_date"),
    F.col("f.customer_id"),
    F.col("f.product"),
    F.col("f.amount"),
    F.col("f.event_timestamp"),

    # Surrogate key: use the real one if found, otherwise placeholder
    F.coalesce(F.col("d.customer_sk"), F.lit(PLACEHOLDER_SK)).alias("customer_sk"),

    # Denormalized dimension attributes (NULL if late)
    F.col("d.customer_name"),
    F.col("d.city"),
    F.col("d.tier"),

    # Reconciliation flag
    F.when(F.col("d.customer_sk").isNull(), F.lit(True))
     .otherwise(F.lit(False))
     .alias("is_late_dimension"),

    # Timestamp for tracking when the record was loaded
    F.current_timestamp().alias("loaded_at"),
)

# Show the results — notice C777 and C888 have placeholder SK = -1
print("🔍 Enriched Facts with Dimension Lookup Results:")
facts_with_sk.select("order_id", "customer_id", "customer_sk", "customer_name", "is_late_dimension").show(truncate=False)

Output:

+----------+-----------+-----------+-------------+------------------+
|order_id  |customer_id|customer_sk|customer_name|is_late_dimension |
+----------+-----------+-----------+-------------+------------------+
|ORD-90001 |C101       |501        |Amit Kumar   |false             |
|ORD-90002 |C102       |502        |Priya Rao    |false             |
|ORD-90003 |C777       |-1         |null         |true              |  ← LATE
|ORD-90004 |C888       |-1         |null         |true              |  ← LATE
|ORD-90005 |C103       |503        |Mukesh Sen   |false             |
+----------+-----------+-----------+-------------+------------------+

Step 5: Write Phase 1 Results — Dual Write Strategy

For TB-scale pipelines, we write to two destinations:

1. Snowflake fact table: All records (including placeholders) go into the warehouse immediately so dashboards aren't blocked.
2. S3 late-dimension staging area: Only the flagged records are parked in a separate S3 path for later reconciliation.

# -------------------------------------------------------
# WRITE 1: All facts → Snowflake (including placeholder SKs)
# This ensures dashboards have today's data immediately
# -------------------------------------------------------
snowflake_write_df = facts_with_sk.drop("is_late_dimension")

snowflake_write_df.write \
    .format("snowflake") \
    .options(**SNOWFLAKE_OPTIONS) \
    .option("dbtable", "FACT_ORDERS") \
    .mode("append") \
    .save()

total_loaded = facts_with_sk.count()
late_count = facts_with_sk.filter("is_late_dimension = true").count()
print(f"✅ Phase 1 complete: {total_loaded:,} facts written to Snowflake ({late_count:,} with placeholder SK)")

# -------------------------------------------------------
# WRITE 2: Only late-dimension records → S3 staging for reconciliation
# Partitioned by order date for efficient re-reads
# -------------------------------------------------------
LATE_STAGING_PATH = "s3://datalake-staging/late-dimensions/fact_orders/"

late_records_df = facts_with_sk.filter("is_late_dimension = true")

late_records_df.write \
    .mode("append") \
    .partitionBy("order_date") \
    .parquet(LATE_STAGING_PATH)

print(f"📁 {late_count:,} late-dimension records staged at {LATE_STAGING_PATH}")

Step 6: Phase 2 — Reconciliation Job (Runs After Dimension Arrives)

Hours later, the late customer dimension feed finally lands in S3. A scheduled reconciliation job (triggered by Airflow, EventBridge, or a cron) picks up the staged records and attempts the lookup again.

# -------------------------------------------------------
# PHASE 2: RECONCILIATION JOB
# This runs as a separate Spark application, triggered after
# the late dimension feed has been loaded into Snowflake
# -------------------------------------------------------

def reconcile_late_dimensions(spark, snowflake_options, late_staging_path, lookback_days=7):
    """
    Reconcile fact records that were loaded with placeholder dimension keys.
    Re-attempts the dimension lookup and updates Snowflake with correct SKs.
    """

    # 1. Read all unreconciled staged records (scan last N days)
    print(f"🔄 Starting reconciliation — scanning last {lookback_days} days...")

    staged_df = spark.read.parquet(late_staging_path)

    pending_count = staged_df.count()
    print(f"📥 Found {pending_count:,} records pending reconciliation")

    if pending_count == 0:
        print("✅ Nothing to reconcile. Exiting.")
        return

    # 2. Read the CURRENT dimension from Snowflake (now includes late arrivals)
    dim_df = spark.read \
        .format("snowflake") \
        .options(**snowflake_options) \
        .option("dbtable", "DIM_CUSTOMER") \
        .load() \
        .filter(F.col("is_current") == True) \
        .select(
            F.col("surrogate_key").alias("resolved_sk"),
            F.col("customer_id"),
            F.col("customer_name").alias("resolved_name"),
            F.col("city").alias("resolved_city"),
            F.col("tier").alias("resolved_tier"),
        )

    dim_df.cache()
    print(f"📦 Dimension reloaded: {dim_df.count():,} active records")

    # 3. Re-attempt the dimension lookup
    reconciled_df = staged_df.alias("s").join(
        dim_df.alias("d"),
        on=F.col("s.customer_id") == F.col("d.customer_id"),
        how="left"
    )

    # Split into resolved and still-unresolved
    resolved_df = reconciled_df.filter(F.col("d.resolved_sk").isNotNull())
    still_pending_df = reconciled_df.filter(F.col("d.resolved_sk").isNull())

    resolved_count = resolved_df.count()
    still_pending_count = still_pending_df.count()
    print(f"✅ Resolved: {resolved_count:,} records")
    print(f"⏳ Still pending: {still_pending_count:,} records")

    # 4. Build the UPDATE statements for Snowflake
    #    We generate a temporary table of corrections and use Snowflake MERGE
    if resolved_count > 0:
        corrections_df = resolved_df.select(
            F.col("s.order_id"),
            F.col("d.resolved_sk").alias("customer_sk"),
            F.col("d.resolved_name").alias("customer_name"),
            F.col("d.resolved_city").alias("city"),
            F.col("d.resolved_tier").alias("tier"),
            F.current_timestamp().alias("reconciled_at"),
        )

        # Write corrections to a Snowflake staging table
        corrections_df.write \
            .format("snowflake") \
            .options(**snowflake_options) \
            .option("dbtable", "FACT_ORDERS_CORRECTIONS") \
            .mode("overwrite") \
            .save()

        # Execute the MERGE in Snowflake using a SQL pushdown
        merge_sql = """
            MERGE INTO FACT_ORDERS AS target
            USING FACT_ORDERS_CORRECTIONS AS source
            ON target.order_id = source.order_id
            WHEN MATCHED AND target.customer_sk = -1 THEN UPDATE SET
                target.customer_sk    = source.customer_sk,
                target.customer_name  = source.customer_name,
                target.city           = source.city,
                target.tier           = source.tier
        """

        # Execute via Snowflake JDBC (using the Spark Snowflake Utils)
        from pyspark._jvm import net  # noqa
        spark._jvm.net.snowflake.spark.snowflake.Utils.runQuery(
            snowflake_options, merge_sql
        )

        print(f"✅ Snowflake MERGE executed: {resolved_count:,} fact records updated with correct SKs")

    # 5. Update the staging area — keep only still-pending records
    if still_pending_count > 0:
        still_pending_df.select("s.*").write \
            .mode("overwrite") \
            .partitionBy("order_date") \
            .parquet(late_staging_path)
        print(f"📁 Staging area updated: {still_pending_count:,} records still pending")
    else:
        # Clear the staging area entirely
        spark.createDataFrame([], staged_df.schema).write \
            .mode("overwrite") \
            .parquet(late_staging_path)
        print("🧹 Staging area cleared — all records reconciled!")

    return resolved_count, still_pending_count


# -------------------------------------------------------
# Execute the reconciliation
# -------------------------------------------------------
reconcile_late_dimensions(spark, SNOWFLAKE_OPTIONS, LATE_STAGING_PATH, lookback_days=7)

Step 7: Verify in Snowflake After Reconciliation

# -------------------------------------------------------
# Verification: Read back the corrected fact records from Snowflake
# -------------------------------------------------------
verification_df = spark.read \
    .format("snowflake") \
    .options(**SNOWFLAKE_OPTIONS) \
    .option("query", """
        SELECT order_id, customer_id, customer_sk, customer_name, city
        FROM FACT_ORDERS
        WHERE order_date = '2026-05-31'
        ORDER BY order_id
    """) \
    .load()

print("🔎 Snowflake Fact Table — After Reconciliation:")
verification_df.show(truncate=False)

# Check no placeholder SKs remain for today
placeholders_remaining = verification_df.filter("customer_sk = -1").count()
print(f"📊 Placeholder SKs remaining: {placeholders_remaining}")

Expected Output After Reconciliation:

+----------+-----------+-----------+-------------+---------+
|order_id  |customer_id|customer_sk|customer_name|city     |
+----------+-----------+-----------+-------------+---------+
|ORD-90001 |C101       |501        |Amit Kumar   |Mumbai   |  ← Was always correct
|ORD-90002 |C102       |502        |Priya Rao    |Pune     |  ← Was always correct
|ORD-90003 |C777       |510        |Ravi Verma   |Jaipur   |  ← RECONCILED! Was -1
|ORD-90004 |C888       |511        |Sunita Devi  |Lucknow  |  ← RECONCILED! Was -1
|ORD-90005 |C103       |503        |Mukesh Sen   |Delhi    |  ← Was always correct
+----------+-----------+-----------+-------------+---------+

TB-Scale Performance Optimizations

When dealing with 2.5 TB of daily facts and 50 GB of dimension data, these optimizations are critical:

1. Broadcast the Dimension Table

At 50 GB the dimension is too large for a default broadcast (10 MB limit). But since it's much smaller than the fact table, force a broadcast join:

# Force broadcast the 50 GB dimension (ensure executor memory can handle it)
from pyspark.sql.functions import broadcast

enriched_df = facts_df.join(
    broadcast(dim_customer_df),
    on="customer_id",
    how="left"
)

Requires --executor-memory 24g+ and --conf spark.sql.autoBroadcastJoinThreshold=55000000000 (55 GB).

2. Partition the Staging Area by Date

Late records are staged with partitionBy("order_date"). The reconciliation job only needs to scan recent partitions, not the entire history.

3. Snowflake Write Parallelism

Use the Snowflake Spark connector's parallelism option and an XL warehouse for bulk writes:

.option("parallelism", "16")          # 16 parallel COPY INTO streams
.option("sfWarehouse", "ETL_WH_XL")  # XL = 128 nodes for fast ingestion

4. Coalesce Before Snowflake Write

Writing millions of tiny files is slow. Coalesce the DataFrame before writing:

facts_with_sk.coalesce(200).write.format("snowflake")...

Follow-Up Questions & Answers

Q1: Why not just skip the fact records where dimension is missing and process them later?

A: At TB scale, "skipping" means buffering potentially millions of fact records indefinitely. This creates several problems:

• Data completeness: Downstream dashboards and reports show incomplete numbers. Finance teams see revenue mismatches.
• Ordering guarantees: If you skip records now and insert them days later, they'll appear as "new" events in time-series dashboards, causing false spikes.
• Complexity: Managing a retry queue of skipped records across days is operationally expensive.

The placeholder approach ensures 100% of facts are loaded on time — dashboards show complete transaction counts, and only the dimension attributes are temporarily missing.

Q2: What goes into the "Unknown" dimension row (SK = -1)?

A: Create a dedicated row in every dimension table with surrogate_key = -1:

INSERT INTO DIM_CUSTOMER (surrogate_key, customer_id, customer_name, city, tier, is_current)
VALUES (-1, 'UNKNOWN', 'Unknown Customer', 'Unknown', 'Unknown', true);

This ensures that any fact record with customer_sk = -1 still has a valid foreign key in Snowflake, and reports can group these under "Unknown Customer" instead of showing NULLs or broken joins.

Q3: How do you monitor reconciliation health in production?

A: Track these metrics daily and alert on anomalies:

# Daily reconciliation metrics
metrics = {
    "date": TODAY,
    "total_facts_loaded": total_loaded,
    "late_dimension_count": late_count,
    "late_percentage": round((late_count / total_loaded) * 100, 2),
    "reconciled_today": resolved_count,
    "still_pending_total": still_pending_count,
    "oldest_pending_date": oldest_pending_order_date,
}

# Alert if late percentage exceeds threshold
if metrics["late_percentage"] > 5.0:
    send_alert(f"⚠️ Late dimension rate is {metrics['late_percentage']}% — exceeds 5% threshold!")

# Alert if records are pending for more than 3 days
if metrics["oldest_pending_date"] < (TODAY - timedelta(days=3)):
    send_alert(f"🚨 Records pending reconciliation since {metrics['oldest_pending_date']}!")

Q4: How does this interact with SCD-2 dimensions?

A: When using SCD-2 dimensions, the reconciliation lookup must account for time-validity. Instead of just matching on customer_id, you must match on customer_id WHERE the fact's event_timestamp falls within the dimension's start_date and end_date range:

reconciled_df = staged_df.alias("s").join(
    dim_df.alias("d"),
    on=(
        (F.col("s.customer_id") == F.col("d.customer_id")) &
        (F.col("s.event_timestamp") >= F.col("d.start_date")) &
        (F.col("s.event_timestamp") < F.col("d.end_date"))
    ),
    how="left"
)

This ensures the fact is linked to the correct historical version of the dimension record that was active at the time the fact event occurred.

Q5: What if the dimension NEVER arrives?

A: Set a maximum pending threshold (e.g., 7 days). If a record has been pending reconciliation for longer than the threshold:

1. Escalate: Send an alert to the source system team to investigate the missing dimension feed.
2. Default: After the threshold, permanently assign the record to the "Unknown" dimension row and stop retrying.
3. Audit log: Write the permanently unresolved records to an audit table for compliance and investigation.

STALE_THRESHOLD_DAYS = 7

stale_df = staged_df.filter(
    F.col("order_date") < F.date_sub(F.current_date(), STALE_THRESHOLD_DAYS)
)

if stale_df.count() > 0:
    print(f"🚨 {stale_df.count()} records have been pending for > {STALE_THRESHOLD_DAYS} days. Marking as permanently unresolved.")
    # Write to audit table and remove from staging

Sub-Scenarios

Sub-Scenario A: Multiple Dimensions Arriving Late Simultaneously

Situation: The fact table references 5 dimensions (customer, product, store, campaign, employee). On a given day, both the product feed and the store feed are late.

Fix: Track placeholder SKs independently per dimension:

facts_with_sk = facts_df \
    .withColumn("customer_sk", F.coalesce(dim_cust_lookup, F.lit(-1))) \
    .withColumn("product_sk",  F.coalesce(dim_prod_lookup, F.lit(-1))) \
    .withColumn("store_sk",    F.coalesce(dim_store_lookup, F.lit(-1))) \
    .withColumn("is_late_customer", F.col("customer_sk") == -1) \
    .withColumn("is_late_product",  F.col("product_sk") == -1) \
    .withColumn("is_late_store",    F.col("store_sk") == -1)

# Stage only the columns relevant to each late dimension
# Reconciliation jobs run independently per dimension

Sub-Scenario B: Late Dimension Arrives in Partial Batches

Situation: The customer dimension feed arrives in two parts — 80% of records at 8 PM and the remaining 20% at 2 AM the next day.

Fix: The reconciliation job is designed to be idempotent and incremental. Run it after each partial batch arrival:

Run 1 (8 PM):  Resolves 80% of pending records
Run 2 (2 AM):  Resolves the remaining 20%
Run 3 (6 AM):  Finds 0 pending records → exits cleanly

Each run only updates records that still have customer_sk = -1 in Snowflake, so running it multiple times is safe.

Sub-Scenario C: Reconciliation at Extreme Scale (10 TB+ Facts)

Situation: Your fact table is 10 TB per day and the reconciliation full-scan is too expensive.

Fix: Avoid scanning the entire fact table. Use the S3 staging area as the only input to the reconciliation job:

❌ Bad:  Read all 10 TB from Snowflake → filter for SK = -1 → reconcile → write back
✅ Good: Read only the staged late records from S3 (typically < 1% of daily volume) → reconcile → MERGE into Snowflake

Our implementation already follows this pattern — the reconciliation job reads only from LATE_STAGING_PATH, not from the full Snowflake fact table.