Pipeline Compaction and Merge Strategies: Managing Small Files, Partition Rewrites, and Storage Optimization in Production Data Lakes

You're three months into production with a streaming ingestion pipeline that lands Parquet files into S3. The pipeline is healthy — data is arriving, quality checks pass, dashboards are green. Then your Athena queries start taking 45 seconds instead of 4. Your Spark jobs are spawning 50,000 tasks for a dataset that should need 200. Your S3 bill climbs 30% even though raw data volume barely changed. You check the partition directory and find it: 180,000 files averaging 8 KB each. You have a small file problem, and it's quietly eating your infrastructure alive.

This is one of the most common, most expensive, and most underestimated failure modes in production data lakes. It doesn't announce itself with an error. It creeps in over weeks as high-frequency writes accumulate, as micro-batch windows land dozens of tiny files per partition per hour, as failed jobs leave behind partial outputs, as watermark-based reprocessing generates correction files alongside originals. By the time you notice it, you're already deep in the hole — and digging out requires more than just running a one-time cleanup job.

By the end of this lesson, you'll understand not just how to compact files but when, why, and at what cost. You'll be able to design a compaction strategy that fits your pipeline's latency requirements, implement partition-aware merge jobs that don't disrupt downstream consumers, and make principled decisions about table format trade-offs.

What you'll learn:

• Why small files degrade performance at multiple layers of the stack and how to quantify the impact
• The four main compaction strategies — inline, deferred, scheduled, and triggered — and when each is appropriate
• How to implement partition-aware compaction jobs in PySpark that handle edge cases correctly
• How Delta Lake, Apache Iceberg, and Apache Hudi approach the small file problem natively, and what each approach costs you
• How to orchestrate compaction safely alongside live pipelines without causing data loss or query inconsistency
• How to tune target file sizes, partition rewrite strategies, and cleanup policies for your specific workload

Prerequisites

You should be comfortable writing PySpark jobs, understand how partitioned data lakes are structured on object storage, and have worked with at least one table format (Parquet on S3, Delta Lake, or Iceberg). Familiarity with incremental loading patterns and partitioning strategies for pipeline output will give you helpful context, though we'll briefly recap the relevant pieces here.

---

Why Small Files Are a Systemic Problem, Not Just a Storage Annoyance

Before designing a solution, you need to understand exactly what breaks and why. Small files hurt you in at least four distinct places in the stack.

The Name Node / Metadata Layer

In HDFS-based systems, the NameNode stores one metadata object per file — path, permissions, block locations, timestamps. The NameNode heap is finite. At scale, a cluster managing 100 million small files will exhaust NameNode memory before it runs out of storage capacity. In cloud object stores like S3 or GCS, you don't have a NameNode, but you do have API rate limits. Listing a directory with 500,000 files requires hundreds of LIST API calls, each with latency and cost. When a Spark job runs sc.textFile("s3://bucket/events/date=2024-01-15/"), it has to enumerate every file in that prefix before it can plan execution. With tens of thousands of files, this listing phase alone can take minutes.

The Task Scheduling Overhead

Spark creates one or more tasks per file (or per Parquet row group). If you have 50,000 files averaging 8 KB each, Spark creates approximately 50,000 tasks. Each task carries scheduling overhead: serialization of the task descriptor, network round-trips to executors, JVM thread startup, metadata reads. For tiny files, this overhead dwarfs the actual compute time. You end up with a cluster burning CPU spinning up tasks that each read 8 KB and complete in milliseconds. Utilization looks high but throughput is low — the worst of both worlds.

Compression Efficiency

Parquet and ORC are column-oriented formats that compress data across rows within a row group. A typical row group target is 128 MB, which gives the compression algorithm enough context to find patterns. A 10 KB Parquet file might have a single row group with 200 rows. At that scale, dictionary encoding is almost useless, delta encoding has no room to work, and the file header/footer overhead represents a significant fraction of the total file size. You'll see compression ratios of 1.2x where a properly-sized file would achieve 5x or 8x. That directly inflates storage costs and scan I/O.

Predicate Pushdown and Statistics

Query engines like Presto, Trino, and Athena rely on Parquet file-level statistics (min/max values per column, null counts) to skip files that can't match a query predicate. This works beautifully when one file covers a meaningful range of values. With 50,000 small files, each covering a tiny slice of data, the statistics become useless for pruning — the engine has to open every file anyway. You've lost one of Parquet's most valuable features.

---

Quantifying Your Small File Problem

Before you can fix anything, measure it. Here's a PySpark diagnostic job that profiles file size distribution across your partition hierarchy:

import boto3
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, LongType

spark = SparkSession.builder.appName("file-size-profiler").getOrCreate()

def list_files_recursive(bucket: str, prefix: str) -> list[dict]:
    """
    Walk an S3 prefix and collect file metadata.
    Returns a list of dicts with path, size, and partition info.
    """
    s3 = boto3.client("s3")
    paginator = s3.get_paginator("list_objects_v2")
    
    results = []
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get("Contents", []):
            key = obj["Key"]
            if key.endswith(".parquet") or key.endswith(".snappy.parquet"):
                # Extract partition path from key
                parts = key.split("/")
                partition_path = "/".join(p for p in parts[:-1] if "=" in p)
                results.append({
                    "path": f"s3://{bucket}/{key}",
                    "partition": partition_path,
                    "size_bytes": obj["Size"],
                    "last_modified": obj["LastModified"].isoformat()
                })
    return results

# Collect metadata
raw_files = list_files_recursive("your-data-lake-bucket", "events/")

schema = StructType([
    StructField("path", StringType()),
    StructField("partition", StringType()),
    StructField("size_bytes", LongType()),
    StructField("last_modified", StringType()),
])

df = spark.createDataFrame(raw_files, schema=schema)

# Profile by partition
partition_profile = df.groupBy("partition").agg(
    F.count("*").alias("file_count"),
    F.sum("size_bytes").alias("total_bytes"),
    F.avg("size_bytes").alias("avg_file_bytes"),
    F.min("size_bytes").alias("min_file_bytes"),
    F.max("size_bytes").alias("max_file_bytes"),
    F.percentile_approx("size_bytes", 0.5).alias("median_file_bytes"),
    F.percentile_approx("size_bytes", 0.95).alias("p95_file_bytes"),
).withColumn(
    "avg_file_mb", F.round(F.col("avg_file_bytes") / 1048576, 2)
).withColumn(
    "total_gb", F.round(F.col("total_bytes") / 1073741824, 3)
).withColumn(
    "compaction_priority",
    F.when(
        (F.col("file_count") > 100) & (F.col("avg_file_bytes") < 10_000_000), 
        "HIGH"
    ).when(
        (F.col("file_count") > 50) & (F.col("avg_file_bytes") < 50_000_000), 
        "MEDIUM"
    ).otherwise("LOW")
).orderBy("file_count", ascending=False)

partition_profile.show(50, truncate=False)

# Summary statistics
total_files = df.count()
files_under_10mb = df.filter(F.col("size_bytes") < 10_000_000).count()
print(f"Total files: {total_files:,}")
print(f"Files under 10 MB: {files_under_10mb:,} ({100 * files_under_10mb / total_files:.1f}%)")
print(f"Fragmentation ratio: {files_under_10mb / total_files:.3f}")

Run this weekly as a monitoring job and alert when any partition exceeds 100 files with an average size below 10 MB. Catching the problem early is vastly cheaper than fixing it at 100,000 files.

---

The Four Compaction Strategies

There's no universal answer to "when should I compact?" The right strategy depends on your latency requirements, query patterns, and the operational complexity you can absorb. Let's work through each approach honestly.

Strategy 1: Inline Compaction (Write-Time Repartitioning)

The simplest approach: at the end of each pipeline run, before writing to the final output path, repartition the DataFrame to hit your target file size.

from pyspark.sql import DataFrame
import math

def write_with_target_file_size(
    df: DataFrame,
    output_path: str,
    partition_cols: list[str],
    target_file_size_mb: int = 128,
    compression: str = "snappy"
) -> None:
    """
    Repartition a DataFrame to approximate target file sizes before writing.
    
    Estimates row size from a sample, computes ideal partition count,
    and writes with that count rather than letting Spark default to
    shuffle partition count (typically 200).
    """
    # Sample to estimate average row size in bytes
    sample_fraction = min(0.1, 10_000 / max(df.count(), 1))
    sample = df.sample(fraction=sample_fraction, seed=42)
    
    # Serialize sample to estimate byte size
    # This uses Spark's internal row serializer as a proxy
    sample_size_bytes = sample.rdd.map(lambda row: len(str(row))).sum()
    sample_row_count = sample.count()
    
    if sample_row_count == 0:
        df.write.partitionBy(*partition_cols).parquet(output_path, compression=compression)
        return
    
    avg_row_bytes = sample_size_bytes / sample_row_count
    total_rows = df.count()
    total_estimated_bytes = avg_row_bytes * total_rows
    
    # Account for compression: Parquet + Snappy typically achieves 4-8x
    # Use conservative 4x estimate to avoid over-compacting
    compression_factor = 4.0
    compressed_bytes = total_estimated_bytes / compression_factor
    
    target_bytes = target_file_size_mb * 1024 * 1024
    ideal_partitions = max(1, math.ceil(compressed_bytes / target_bytes))
    
    print(f"Estimated compressed size: {compressed_bytes / 1e9:.2f} GB")
    print(f"Target file size: {target_file_size_mb} MB")
    print(f"Writing with {ideal_partitions} partitions")
    
    (
        df.repartition(ideal_partitions, *partition_cols)
        .write
        .mode("overwrite")
        .partitionBy(*partition_cols)
        .option("compression", compression)
        .parquet(output_path)
    )

This works well for batch pipelines where you control the entire write. The problem is that it doesn't compose well with incremental patterns. If you're appending to an existing partitioned dataset, you can't just repartition the new data — you have to think about how it interacts with what's already on disk.

Strategy 2: Deferred Compaction (Separate Merge Job)

This is the pattern you'll use most often in production. The ingestion pipeline writes as fast as possible with minimal concern for file size, and a separate compaction job runs on a schedule to merge small files. This decouples write throughput from storage efficiency.

from pyspark.sql import SparkSession, DataFrame
from pyspark.sql import functions as F
from typing import Optional
import logging

logger = logging.getLogger(__name__)

def compact_partition(
    spark: SparkSession,
    table_path: str,
    partition_spec: dict[str, str],
    target_file_size_mb: int = 256,
    min_file_count_threshold: int = 10,
    dry_run: bool = False,
) -> dict:
    """
    Compact all files in a specific partition into target-sized files.
    
    Args:
        table_path: Root path of the table (e.g., s3://bucket/events)
        partition_spec: Dict of partition column -> value 
                        (e.g., {"date": "2024-01-15", "region": "us-east"})
        target_file_size_mb: Target output file size in megabytes
        min_file_count_threshold: Skip compaction if file count is below this
        dry_run: If True, analyze but don't write
    
    Returns:
        Dict with compaction metrics
    """
    # Build partition path
    partition_path = "/".join(
        f"{col}={val}" for col, val in sorted(partition_spec.items())
    )
    full_path = f"{table_path}/{partition_path}"
    
    # Read existing data in this partition
    try:
        existing_df = spark.read.parquet(full_path)
    except Exception as e:
        logger.warning(f"Could not read partition {full_path}: {e}")
        return {"status": "skipped", "reason": "unreadable"}
    
    # Get file count for this partition
    # Using Spark's input file tracking
    files_df = existing_df.select(F.input_file_name().alias("_file")).distinct()
    file_count = files_df.count()
    
    if file_count < min_file_count_threshold:
        logger.info(f"Partition {partition_path} has {file_count} files, skipping")
        return {"status": "skipped", "file_count": file_count, "reason": "below_threshold"}
    
    total_rows = existing_df.count()
    
    if dry_run:
        return {
            "status": "dry_run",
            "partition": partition_path,
            "file_count": file_count,
            "total_rows": total_rows,
        }
    
    # Compute optimal partition count
    # We'll write to a temp path first to avoid corrupting the original
    temp_path = f"{table_path}/_compaction_temp/{partition_path}"
    
    # Estimate compressed output size
    # Read sample to get uncompressed estimate, apply compression ratio
    sample_df = existing_df.sample(fraction=0.01, seed=42)
    
    # Use Spark's plan statistics as a better estimate
    existing_df.cache()
    
    target_bytes = target_file_size_mb * 1024 * 1024
    # Conservative estimate: assume 5x compression for columnar format
    estimated_output_bytes = (total_rows * 200) / 5  # 200 bytes/row uncompressed estimate
    optimal_partitions = max(1, round(estimated_output_bytes / target_bytes))
    
    logger.info(
        f"Compacting {partition_path}: {file_count} files -> "
        f"~{optimal_partitions} files, {total_rows:,} rows"
    )
    
    # Write compacted data to temp location
    (
        existing_df
        .repartition(optimal_partitions)
        .write
        .mode("overwrite")
        .option("compression", "snappy")
        .parquet(temp_path)
    )
    
    existing_df.unpersist()
    
    # Atomic swap: move temp to final
    # In practice on S3, you'd use a rename/copy+delete pattern
    # or an atomic operation from your catalog (Delta OPTIMIZE, etc.)
    _atomic_swap(spark, temp_path, full_path)
    
    # Verify the result
    compacted_df = spark.read.parquet(full_path)
    new_files_df = compacted_df.select(F.input_file_name().alias("_file")).distinct()
    new_file_count = new_files_df.count()
    new_row_count = compacted_df.count()
    
    if new_row_count != total_rows:
        raise ValueError(
            f"Row count mismatch after compaction: "
            f"before={total_rows}, after={new_row_count}"
        )
    
    return {
        "status": "success",
        "partition": partition_path,
        "files_before": file_count,
        "files_after": new_file_count,
        "rows": total_rows,
        "reduction_ratio": round(file_count / new_file_count, 2),
    }


def _atomic_swap(spark: SparkSession, src: str, dst: str) -> None:
    """
    Move compacted files from temp to final location.
    On S3, 'rename' is copy + delete, so this isn't truly atomic.
    For true atomicity, use a table format (Delta, Iceberg, Hudi).
    """
    hadoop_conf = spark._jsc.hadoopConfiguration()
    fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(
        spark._jvm.java.net.URI.create(src),
        hadoop_conf
    )
    src_path = spark._jvm.org.apache.hadoop.fs.Path(src)
    dst_path = spark._jvm.org.apache.hadoop.fs.Path(dst)
    
    # Delete destination, then rename source to destination
    fs.delete(dst_path, True)  # recursive delete
    fs.rename(src_path, dst_path)

The critical piece you'll notice is the _atomic_swap function — and its comment about non-atomicity on S3. This is one of the most important operational hazards in bare-Parquet compaction.

Strategy 3: Scheduled Triggered Compaction

Rather than compacting on a fixed schedule, you trigger compaction based on observed conditions — file count crossing a threshold, total partition size in small files exceeding a budget, or time since last compaction. This is more efficient because you don't waste compute compacting partitions that don't need it.

def build_compaction_candidates(
    spark: SparkSession,
    profile_table_path: str,
    max_avg_file_mb: float = 64.0,
    min_files: int = 20,
    max_age_days: int = 7,
) -> list[dict]:
    """
    Query the file profile table to identify partitions needing compaction.
    Prioritize by file count (most fragmented first).
    """
    from datetime import datetime, timedelta
    
    cutoff_date = (datetime.utcnow() - timedelta(days=max_age_days)).strftime("%Y-%m-%d")
    
    candidates = (
        spark.read.parquet(profile_table_path)
        .filter(
            (F.col("avg_file_mb") < max_avg_file_mb) &
            (F.col("file_count") >= min_files) &
            (F.col("profile_date") >= cutoff_date)
        )
        .orderBy(F.col("file_count").desc())
        .select("partition", "file_count", "avg_file_mb", "total_gb")
        .collect()
    )
    
    return [row.asDict() for row in candidates]

Wire this into your Airflow orchestration by running the profiler daily, storing results, and triggering compaction tasks only for partitions that exceed thresholds.

Strategy 4: Table Format Native Compaction

If you're using Delta Lake, Apache Iceberg, or Apache Hudi, you get compaction built into the table format. This is the right long-term answer for most teams.

---

Compaction in Delta Lake, Iceberg, and Hudi

Each of the major open table formats approaches compaction differently, and the differences matter for your operational model.

Delta Lake: OPTIMIZE and Z-ORDER

Delta Lake's OPTIMIZE command rewrites files in a partition to approach your target file size (default 1 GB, tunable with spark.databricks.delta.optimize.maxFileSize). What makes Delta's approach safe is its transaction log: OPTIMIZE is a committed transaction, so concurrent readers see either the old files or the new files — never a mixture of old and partially-written new files.

from delta.tables import DeltaTable

# Basic optimization of a specific partition
spark.sql("""
    OPTIMIZE events
    WHERE date >= '2024-01-01' AND date <= '2024-01-31'
""")

# Z-ORDER co-locates related data within files to improve
# predicate pushdown on high-cardinality columns
spark.sql("""
    OPTIMIZE events
    WHERE date = '2024-01-15'
    ZORDER BY (user_id, event_type)
""")

# Programmatic access via Delta API
delta_table = DeltaTable.forPath(spark, "s3://bucket/events")
delta_table.optimize().where("date = '2024-01-15'").executeCompaction()

After OPTIMIZE, the old small files aren't deleted immediately — they're marked as "removed" in the transaction log but the physical files remain. This is important for time travel. You control cleanup with VACUUM:

# Remove files no longer referenced by the transaction log
# that are older than the retention period (default 7 days)
spark.sql("VACUUM events RETAIN 168 HOURS")

# DeltaTable API
delta_table.vacuum(retentionHours=168)

Apache Iceberg: Rewrites via Spark Procedures

Iceberg's compaction model is more flexible and arguably more sophisticated. It separates data file compaction (merging small files) from manifest compaction (merging the metadata files that list data files), which matters at scale because the metadata layer can itself become a small file problem.

# Rewrite data files in Iceberg
spark.sql("""
    CALL catalog.system.rewrite_data_files(
        table => 'db.events',
        strategy => 'binpack',
        options => map(
            'target-file-size-bytes', '268435456',
            'min-file-size-bytes', '67108864',
            'max-file-size-bytes', '536870912',
            'min-input-files', '5',
            'rewrite-all', 'false'
        ),
        where => 'date = ''2024-01-15'''
    )
""")

# Rewrite manifests - often overlooked but important
spark.sql("""
    CALL catalog.system.rewrite_manifests(
        table => 'db.events',
        use_caching => true
    )
""")

# Expire old snapshots (equivalent to Delta's VACUUM)
spark.sql("""
    CALL catalog.system.expire_snapshots(
        table => 'db.events',
        older_than => TIMESTAMP '2024-01-01 00:00:00',
        retain_last => 5
    )
""")

Iceberg supports three compaction strategies: binpack (default, just packs files to fill target size), sort (rewrites files in sorted order by specified columns), and zorder (Z-order curve layout, similar to Delta's ZORDER). The sort and zorder strategies are significantly more expensive computationally but produce better scan performance for selective queries.

Apache Hudi: Compaction as a First-Class Concept

Hudi distinguishes between two table types: Copy-on-Write (CoW) and Merge-on-Read (MoR). This distinction fundamentally changes how compaction works.

In CoW tables, writes always produce full file rewrites — there are no small delta files because every write produces a complete new version of affected files. This gives you excellent read performance at the cost of write amplification.

In MoR tables, writes land as small log files (delta files) alongside base Parquet files. Reads merge base files with log files on the fly. Compaction is the process of materializing those log files back into base files:

# Hudi compaction is typically run as a separate Spark job
# For inline compaction (slower writes, no separate job needed):
hudi_options = {
    "hoodie.compact.inline": "true",
    "hoodie.compact.inline.max.delta.commits": "5",  # Compact after 5 delta commits
    "hoodie.parquet.max.file.size": str(128 * 1024 * 1024),
    "hoodie.parquet.small.file.limit": str(104 * 1024 * 1024),
}

# For async compaction (recommended for production):
# Run the compaction job separately using HoodieCompactor
# hoodie-spark-client --spark-memory 4g \
#   --cmd compact \
#   --base-path s3://bucket/events \
#   --table-name events \
#   --instant-time 20240115120000

---

The Atomic Swap Problem on Object Storage

One of the trickiest aspects of raw-Parquet compaction (without a table format) is making the swap atomic. S3 does not support directory renames — what looks like a rename is actually a copy-then-delete sequence. During the time between the copy completing and the delete finishing, you have two copies of the data, and a reader hitting the directory at the wrong moment could see duplicated records or a mix of old and new files.

The safest approach is to use a swap pattern based on symlinks or partition metadata, but practically speaking, the real solution is to use a table format that provides transactional guarantees. If you must use bare Parquet, here's a safer compaction pattern using partition-level staging:

def safe_parquet_compaction(
    spark: SparkSession,
    table_path: str,
    partition_spec: dict[str, str],
    staging_bucket: str,
) -> None:
    """
    Compaction pattern that minimizes the window of inconsistency.
    
    Strategy:
    1. Write compacted files to a staging location
    2. Verify row counts match
    3. Delete original partition files one by one
    4. Move compacted files into the partition directory
    
    This is still not truly atomic, but minimizes inconsistency window
    and is recoverable at each step.
    """
    partition_suffix = "/".join(f"{k}={v}" for k, v in sorted(partition_spec.items()))
    src_path = f"{table_path}/{partition_suffix}"
    staging_path = f"s3://{staging_bucket}/compaction_staging/{partition_suffix}"
    
    # Step 1: Read original data
    original_df = spark.read.parquet(src_path)
    original_count = original_df.count()
    
    # Step 2: Write compacted to staging
    target_partitions = max(1, round(original_count / 500_000))  # ~500K rows/file
    (
        original_df
        .repartition(target_partitions)
        .write
        .mode("overwrite")
        .parquet(staging_path)
    )
    
    # Step 3: Verify staging
    staged_df = spark.read.parquet(staging_path)
    staged_count = staged_df.count()
    
    if staged_count != original_count:
        # Clean up staging and abort
        _delete_path(spark, staging_path)
        raise ValueError(f"Count mismatch: original={original_count}, staged={staged_count}")
    
    # Step 4: Schema verification
    if set(original_df.columns) != set(staged_df.columns):
        _delete_path(spark, staging_path)
        raise ValueError("Schema mismatch between original and compacted data")
    
    # Step 5: Swap - delete originals, move staged files
    # This is the non-atomic window. Queries during this window may see
    # an empty or partial partition. Accept this or use a table format.
    _delete_path(spark, src_path)
    _move_path(spark, staging_path, src_path)
    
    print(f"Compaction complete: {original_count:,} rows, verified.")

This pattern is viable but fragile. The explicit row count and schema verification before the swap catches most data loss scenarios. For designing truly idempotent pipelines, a table format is the right foundation.

---

Partition Rewrite Strategies for Schema Evolution

Compaction often intersects with another operational challenge: what do you do when you need to rewrite a partition because the schema changed? This is distinct from compaction but uses the same machinery.

The wrong approach is to read the partition, add the new column, and overwrite in place. The right approach — especially when you need to handle schema evolution in production — is to treat every partition rewrite as a full compaction job: read, transform, stage, verify, swap.

def rewrite_partition_with_schema_change(
    spark: SparkSession,
    table_path: str,
    partition_spec: dict[str, str],
    transformation_fn,  # callable: DataFrame -> DataFrame
    verify_fn=None,     # optional callable: (DataFrame, DataFrame) -> bool
) -> dict:
    """
    Rewrite a partition applying a schema transformation.
    
    Example use cases:
    - Adding a derived column to historical partitions
    - Changing column types (e.g., string -> long for IDs)
    - Renaming columns to match new naming conventions
    - Re-partitioning within a partition (changing sort order)
    """
    partition_suffix = "/".join(f"{k}={v}" for k, v in sorted(partition_spec.items()))
    src_path = f"{table_path}/{partition_suffix}"
    
    # Read with schema inference disabled — use explicit schema
    # to catch schema mismatches early
    original_df = spark.read.option("mergeSchema", "false").parquet(src_path)
    original_count = original_df.count()
    
    # Apply transformation
    transformed_df = transformation_fn(original_df)
    transformed_count = transformed_df.count()
    
    # Row count should be preserved for schema-only changes
    if transformed_count != original_count:
        raise ValueError(
            f"Row count changed during transformation: "
            f"{original_count} -> {transformed_count}. "
            f"If this is intentional (deduplication), pass verify_fn=None and handle manually."
        )
    
    # Optional custom verification
    if verify_fn is not None:
        if not verify_fn(original_df, transformed_df):
            raise ValueError("Custom verification failed")
    
    # Write to staging and swap
    staging_path = f"{table_path}/_rewrite_staging/{partition_suffix}"
    optimal_partitions = max(1, round(original_count / 1_000_000))
    
    (
        transformed_df
        .repartition(optimal_partitions)
        .write
        .mode("overwrite")
        .parquet(staging_path)
    )
    
    # Verify staged output
    staged_df = spark.read.parquet(staging_path)
    assert staged_df.count() == original_count, "Staged row count mismatch"
    
    _delete_path(spark, src_path)
    _move_path(spark, staging_path, src_path)
    
    return {
        "partition": partition_suffix,
        "original_count": original_count,
        "new_columns": [c for c in transformed_df.columns if c not in original_df.columns],
        "status": "success"
    }

# Example: Add a derived `hour` column to all historical date partitions
def add_hour_column(df):
    return df.withColumn("hour", F.hour(F.col("event_timestamp")))

# Apply to a range of historical partitions
for date in ["2024-01-01", "2024-01-02", "2024-01-03"]:
    result = rewrite_partition_with_schema_change(
        spark=spark,
        table_path="s3://bucket/events",
        partition_spec={"date": date},
        transformation_fn=add_hour_column,
    )
    print(result)

---

Orchestrating Compaction Alongside Live Pipelines

Compaction jobs don't run in isolation — your ingestion pipeline is probably writing new data while compaction runs. This creates race conditions you need to design around.

The safest pattern is partition-level exclusion: your compaction job should only touch partitions that the ingestion pipeline is not currently writing to. For event-time partitioned data, this typically means compacting partitions more than N hours or days old.

from datetime import datetime, timedelta

def get_compaction_eligible_partitions(
    all_partitions: list[str],
    active_pipeline_lag_hours: int = 3,
    max_lookback_days: int = 90,
) -> list[str]:
    """
    Filter partitions eligible for compaction.
    
    Excludes:
    - Today's partition and recent partitions (ingestion still active)
    - Partitions older than max_lookback_days (diminishing returns)
    - Partitions currently being written by the pipeline
    """
    now = datetime.utcnow()
    cutoff_recent = now - timedelta(hours=active_pipeline_lag_hours)
    cutoff_old = now - timedelta(days=max_lookback_days)
    
    eligible = []
    for partition in all_partitions:
        # Parse date from partition string like "date=2024-01-15"
        try:
            date_str = next(p.split("=")[1] for p in partition.split("/") if p.startswith("date="))
            partition_date = datetime.strptime(date_str, "%Y-%m-%d")
        except (StopIteration, ValueError):
            continue
        
        if cutoff_old <= partition_date <= cutoff_recent:
            eligible.append(partition)
    
    return sorted(eligible, reverse=True)  # Most recent first

In Airflow, you'd run the compaction DAG on an offset schedule from the ingestion DAG — for example, ingestion runs at the top of each hour, compaction runs at :30 past, targeting partitions older than 3 hours. This isn't bulletproof but it's a practical approximation for most use cases.

---

Storage Optimization Beyond Compaction

Compaction solves the file count problem but doesn't automatically solve the storage cost problem. A few additional strategies compound well with compaction.

Tiered Storage and Lifecycle Policies

Most data lakes have a hot/warm/cold access pattern: yesterday's events are queried constantly, last month's events occasionally, last year's events almost never. Configure S3 lifecycle rules to move old partitions to cheaper storage tiers:

{
  "Rules": [
    {
      "ID": "move-old-events-to-ia",
      "Status": "Enabled",
      "Filter": {"Prefix": "events/"},
      "Transitions": [
        {"Days": 30, "StorageClass": "STANDARD_IA"},
        {"Days": 90, "StorageClass": "GLACIER_IR"},
        {"Days": 365, "StorageClass": "DEEP_ARCHIVE"}
      ]
    }
  ]
}

Choosing Compression Wisely

Once you have properly-sized files, the compression codec choice matters. For Parquet and other columnar formats, Snappy is the default because it's fast to decompress. ZSTD (zstandard) offers significantly better compression ratios with comparable decompression speed:

# Snappy: ~3-4x compression, fast decompress
df.write.parquet(path, compression="snappy")

# ZSTD level 3: ~5-7x compression, still fast
df.write.option("parquet.compression", "ZSTD") \
        .option("parquet.compression.codec.level", "3") \
        .parquet(path)

# ZSTD level 9: ~7-10x compression, slower decompress
# Good for cold/archival data
df.write.option("parquet.compression", "ZSTD") \
        .option("parquet.compression.codec.level", "9") \
        .parquet(path)

For archival partitions being accessed rarely but still needing to be queryable, ZSTD level 9 can reduce storage costs by 30-40% compared to Snappy with minimal query performance penalty (query time is dominated by I/O scheduling, not CPU decompression).

Row Group and Page Size Tuning

Inside a Parquet file, row groups are the unit of filtering via column statistics. Larger row groups mean better compression and more effective min/max statistics, but also more memory pressure during reads. For most workloads, 128 MB row groups are reasonable. For high-selectivity analytical queries on wide tables, consider 256 MB:

spark.conf.set("spark.sql.parquet.blockSize", str(256 * 1024 * 1024))  # 256 MB row groups
spark.conf.set("spark.sql.parquet.pageSize", str(1 * 1024 * 1024))     # 1 MB pages

---

Hands-On Exercise

Set up a local simulation of the small file problem and work through compaction with real metrics.

Setup (10 minutes):

Start a local Spark session and generate a realistic fragmented dataset:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *
import random
from datetime import datetime, timedelta

spark = SparkSession.builder \
    .master("local[4]") \
    .config("spark.sql.shuffle.partitions", "8") \
    .appName("compaction-exercise") \
    .getOrCreate()

# Simulate 30 days of micro-batch writes
# Each "micro-batch" writes a tiny file to a date partition
schema = StructType([
    StructField("user_id", LongType()),
    StructField("event_type", StringType()),
    StructField("event_timestamp", TimestampType()),
    StructField("session_id", StringType()),
    StructField("revenue_usd", DoubleType()),
])

base_date = datetime(2024, 1, 1)
output_path = "/tmp/exercise_lake/events"

for day_offset in range(30):
    event_date = base_date + timedelta(days=day_offset)
    date_str = event_date.strftime("%Y-%m-%d")
    
    # Simulate 20 micro-batches per day landing small files
    for batch_num in range(20):
        batch_size = random.randint(500, 2000)  # Small batches
        
        data = [(
            random.randint(1, 1_000_000),
            random.choice(["click", "view", "purchase", "search"]),
            event_date + timedelta(hours=batch_num, minutes=random.randint(0, 59)),
            f"sess_{random.randint(1, 100000):08d}",
            round(random.uniform(0, 200), 2) if random.random() < 0.1 else 0.0
        ) for _ in range(batch_size)]
        
        batch_df = spark.createDataFrame(data, schema=schema)
        
        (batch_df
         .write
         .mode("append")
         .partitionBy("date")  # Note: this won't work without adding date column
         .option("compression", "snappy")
         .parquet(output_path))

print("Dataset generated. Now profile it:")

Exercise Tasks:

1. Profile the dataset: Adapt the file profiling code from earlier to count files per partition and compute average file sizes. Record your baseline metrics.

2. Implement and run compaction: Use the compact_partition function to compact the partition for date=2024-01-01. Measure before and after: file count, total size in bytes, and the time taken to run a query that filters to that partition.

3. Measure query time improvement: Run spark.read.parquet(output_path).filter(F.col("date") == "2024-01-01").agg(F.sum("revenue_usd")).show() before and after compaction. Record query time.

4. Experiment with target file sizes: Run compaction three times with target sizes of 32 MB, 128 MB, and 256 MB. Compare the resulting file counts and query times. Identify the inflection point where larger files stop helping.

5. Simulate the race condition: While a compaction job is running (artificially slowed with time.sleep), append new data to the same partition from a separate process. Observe what happens to row counts if you don't have transactional guarantees.

Expected outcomes: You should see query time drop by 5x-15x after compaction, file count drop from ~20 files to 1-2 files per partition, and compression ratio improve noticeably.

---

Common Mistakes & Troubleshooting

Compacting the Hot Partition

Compacting today's partition while the ingestion pipeline is actively writing to it is a classic mistake. You'll see intermittent duplicate rows (compaction reads the files that exist, new files arrive, compaction's output includes the old data, append adds more) or missing rows (compaction's swap deletes a file that was just written by the pipeline). Fix: always exclude the current day's partition and at least one lag buffer.

Using `coalesce()` Across Partition Boundaries

df.coalesce(1) before a partitionBy() write is a common mistake. This sends all data through a single executor for merging before distributing it to partition writers. It's slower than a proper shuffle-based repartition and can OOM on large datasets. Use repartition(n, *partition_cols) to shuffle data to the right partition and then coalesce within each partition.

Forgetting to VACUUM After Delta OPTIMIZE

Running OPTIMIZE without subsequently running VACUUM leaves all the old small files on disk. Your query performance improves (because the transaction log points to the new large files), but your storage costs don't decrease. Many teams run OPTIMIZE regularly but forget VACUUM, then wonder why their S3 bill hasn't changed. Set a reminder: OPTIMIZE and VACUUM are a pair.

Underestimating Manifest Compaction in Iceberg

In Iceberg, the manifest layer can itself develop small file problems. After thousands of small writes, you'll have thousands of manifest files. Iceberg's query planner has to read all manifests to plan a query. Running rewrite_data_files without also running rewrite_manifests solves the data layer but leaves the metadata layer fragmented. Make both procedures part of your compaction runbook.

Compaction Causing Downstream Inconsistency in Multi-System Pipelines

If you have a pipeline that reads from your data lake and writes derived data to a downstream system (say, a feature store or a cache), compacting the source partition can invalidate any change-detection logic based on file modification timestamps or ETags. The downstream pipeline sees a "changed" file and reprocesses data it already handled, producing duplicates. This is particularly nasty when combined with CDC-based incremental loading patterns. The fix is to track logical checkpoints (record counts, max timestamps) rather than file metadata in downstream pipelines.

Z-ORDER on Low-Cardinality Columns

Z-ORDER works by interleaving multiple high-cardinality sort keys to co-locate records that share values across multiple dimensions. Applying it to low-cardinality columns like event_type (4 values) or region (3 values) provides little benefit because there aren't enough distinct values to create meaningful locality. Z-ORDER pays off on high-cardinality columns like user_id, session_id, or device_id. Profile your query predicates before investing compute in Z-ORDER.

Running Compaction Without Resource Isolation

A compaction job reading and rewriting 500 GB of data on your shared Spark cluster will compete with production query workloads. Compaction is CPU and I/O intensive. If you're not running on dedicated compaction infrastructure (a separate Spark cluster, a Databricks job cluster, or EMR on-demand), schedule compaction during off-peak hours and set resource limits:

# Limit parallelism for compaction to avoid starving queries
spark.conf.set("spark.default.parallelism", "32")       # Reduce from default
spark.conf.set("spark.sql.shuffle.partitions", "32")     # Reduce from 200
spark.conf.set("spark.executor.cores", "4")              # Limit per-executor cores

---

Summary & Next Steps

Small file accumulation is a systemic problem that compounds over time. It degrades performance at multiple layers — name nodes, task scheduling, compression, and predicate pushdown — and the damage is often invisible until it's severe. The key takeaways from this lesson:

On diagnosis: Profile your file size distribution per partition regularly and alert proactively. The fragmentation ratio (small files as a fraction of total files) is a leading indicator. Don't wait for users to report slow queries.

On strategy choice: Inline compaction is simple but doesn't compose with incremental patterns. Deferred compaction gives you decoupled write throughput. Table format native compaction (Delta OPTIMIZE, Iceberg rewrite_data_files, Hudi async compaction) gives you transactional safety that bare-Parquet approaches can't match.

On safety: The atomic swap problem on object storage is real. Row count verification before and after compaction is non-negotiable. Exclude hot partitions from compaction. Understand how downstream consumers detect changes before running large sweeps.

On cost: Compaction, Z-ORDER, and compression improvements all interact. The right sequence is compact first (fix file count), then apply Z-ORDER for high-cardinality query patterns, then tune compression for your access temperature.

For next steps, consider exploring backpressure and throughput tuning to understand why your pipeline is generating small files in the first place — often the root cause is batch window sizing, not compaction lag. If you're running high-frequency incremental pipelines, checkpointing and state management patterns will help you coordinate compaction safely with live ingestion. And if cost optimization is the primary driver, cost attribution and pipeline-level resource optimization covers the full landscape of techniques for reducing cloud spend in data workflows.