Compacting Spatial Iceberg Tables with rewrite_data_files

This guide gives a complete, runnable recipe for compacting a small-file-ridden spatial Iceberg table using the rewrite_data_files procedure with a bounding-box sort order, tuned target file size, and partial-progress safety, then verifying the result through the files metadata table.

Context and prerequisites

Spatial ingestion — streaming GPS pings, per-tile raster footprints, incremental vector loads — produces many small Parquet files, each with a wide bounding-box statistics footprint. Query planning slows because the engine must open and filter thousands of manifests, and bbox-based predicate pushdown prunes poorly when co-located geometries are scattered across files. The fix is a sort-based rewrite that both merges small files toward a target size and physically clusters rows by their bounding box. This is one of the maintenance disciplines automated in lakehouse maintenance automation, and it assumes Apache Iceberg 1.4+ (examples use the 1.9.0 Spark runtime) on Spark 3.5, with a table carrying bbox_minx/miny/maxx/maxy columns computed at ingest.

Complete working solution

The following script builds a Spark session against a REST catalog, checks whether the table is fragmented enough to warrant a rewrite, runs the sort-based rewrite_data_files with spatial sort order, and prints the procedure’s result row (rewritten and added file counts).

from pyspark.sql import SparkSession

def build_spark(app: str = "iceberg-spatial-compaction") -> SparkSession:
    return (
        SparkSession.builder.appName(app)
        .config("spark.jars.packages",
                "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.9.0")
        .config("spark.sql.extensions",
                "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
        .config("spark.sql.catalog.lake", "org.apache.iceberg.spark.SparkCatalog")
        .config("spark.sql.catalog.lake.type", "rest")
        .config("spark.sql.catalog.lake.uri", "http://catalog:8181")
        .config("spark.sql.catalog.lake.warehouse", "s3://lakehouse/wh")
        .config("spark.sql.shuffle.partitions", "64")
        .getOrCreate()
    )

def fragmentation_ratio(spark, table_fqn: str) -> tuple[int, float]:
    """Return (file_count, avg_file_mb) from the files metadata table."""
    row = spark.sql(f"""
        SELECT count(*) AS n,
               avg(file_size_in_bytes) / 1048576 AS avg_mb
        FROM lake.{table_fqn}.files
    """).collect()[0]
    return int(row["n"]), float(row["avg_mb"] or 0.0)

def compact(spark, table_fqn: str,
            target_bytes: int = 134_217_728,   # 128 MB
            min_input_files: int = 5) -> None:
    """Sort-based rewrite clustering rows on the bounding box."""
    result = spark.sql(f"""
        CALL lake.system.rewrite_data_files(
          table => '{table_fqn}',
          strategy => 'sort',
          sort_order => 'bbox_minx ASC NULLS LAST, bbox_miny ASC NULLS LAST',
          options => map(
            'target-file-size-bytes', '{target_bytes}',
            'min-input-files', '{min_input_files}',
            'max-concurrent-file-group-rewrites', '4',
            'partial-progress.enabled', 'true',
            'partial-progress.max-commits', '10',
            'rewrite-job-order', 'files-desc'
          )
        )
    """)
    result.show(truncate=False)

if __name__ == "__main__":
    spark = build_spark()
    table = "gis.telemetry"
    n_before, avg_before = fragmentation_ratio(spark, table)
    print(f"before: {n_before} files, avg {avg_before:.1f} MB")

    # only rewrite when fragmentation actually justifies the compute
    if avg_before < 64.0 or n_before > 200:
        compact(spark, table)
        n_after, avg_after = fragmentation_ratio(spark, table)
        print(f"after:  {n_after} files, avg {avg_after:.1f} MB")
    else:
        print("table already well-sized; skipping rewrite")
    spark.stop()

Step-by-step walkthrough

1. build_spark wires the Iceberg 1.9.0 Spark 3.5 runtime, registers the IcebergSparkSessionExtensions (required for the CALL procedure syntax), and points at a REST catalog named lake. spark.sql.shuffle.partitions is set to 64 to match a modest executor pool — this governs how the sort output is partitioned and therefore how many files the rewrite emits.

2. fragmentation_ratio reads the built-in .files metadata table, which exposes one row per data file with file_size_in_bytes. Computing the average size and count here is cheap (metadata-only) and lets you skip the expensive rewrite when it is not needed.

3. strategy => 'sort' selects a sort-based rewrite rather than the default binpack. Binpack only merges files by size; sort additionally reorders rows. The sort_order clusters rows by bbox_minx then bbox_miny, so geometries near each other in space land in the same file — the property that makes file-level bbox pruning effective.

4. target-file-size-bytes = 128 MB is the size the rewriter aims each output file toward. It is the single most important spatial tuning knob: 128 MB balances scan efficiency against pruning granularity.

5. min-input-files = 5 stops the procedure from rewriting file groups that already have fewer than five files, avoiding pointless churn on nearly-compacted partitions.

6. max-concurrent-file-group-rewrites = 4 bounds parallelism to keep object-storage request rates under the throttling threshold.

7. partial-progress.enabled = true with max-commits = 10 checkpoints the rewrite across up to ten commits, so a failure late in a large rewrite keeps the work already done instead of discarding all of it.

8. rewrite-job-order = ‘files-desc’ processes the most-fragmented file groups first, so the biggest planning wins land earliest even if the job is interrupted.

The gate before compact is the same non-blocking, threshold-driven pattern used across async execution patterns — never pay for a rewrite the table does not need.

Why sort, not binpack

Iceberg offers two rewrite strategies and choosing wrong wastes the entire operation for spatial workloads. Binpack is a pure size operation: it concatenates adjacent small files into target-sized ones without touching row order. It is fast and cheap, but it does nothing for spatial locality — if your small files were each written from a different ingest batch spanning the whole globe, binpacking them produces large files that each still span the whole globe, so a bounding-box filter cannot skip any of them. Sort pays the extra cost of a full shuffle-and-sort so that rows are reordered by the sort key before being written into target-sized files. With bbox_minx, bbox_miny as the key, each output file ends up covering a compact, contiguous slice of space, and its min/max statistics become tight. That tightness is what the query planner reads to skip files. The rule: use binpack only when the table is already spatially partitioned and you just need to defragment within a partition; use sort whenever file-level bbox pruning matters, which for a queryable spatial table is almost always.

The single-dimension sort used here (X then Y) is simpler than a space-filling curve but has a known weakness: it clusters tightly along longitude and only loosely along latitude. When query patterns are truly two-dimensional — bounding-box windows rather than longitude strips — a Z-order or Hilbert interleaving of the bbox columns clusters better in both dimensions at once. That variant, and when it is worth the extra cost, is covered in Z-ordering for geospatial queries; for a table whose queries are dominated by one axis (a north-south transport corridor, say) the linear sort here is both cheaper and sufficient.

Common errors and fixes

Verification

Confirm the rewrite merged files and left the data unchanged. First, per-partition file health from the metadata table:

-- Spark SQL
SELECT partition,
       count(*) AS file_count,
       cast(avg(file_size_in_bytes) / 1048576 AS int) AS avg_mb
FROM lake.gis.telemetry.files
GROUP BY partition
ORDER BY file_count DESC;

Then prove the rewrite was layout-only by comparing row count and aggregate bounding-box extent against the pre-rewrite values — they must be identical:

SELECT count(*) AS n,
       min(bbox_minx) AS xmin, min(bbox_miny) AS ymin,
       max(bbox_maxx) AS xmax, max(bbox_maxy) AS ymax
FROM lake.gis.telemetry;

Inspect the manifests directly to confirm the rewrite consolidated them and that the bbox statistics are now tight. The manifests metadata table lists each manifest with its partition-value bounds and the count of data files it tracks; after a successful sort-rewrite you should see fewer manifests, each referencing more files, with narrower spatial ranges:

-- Spark SQL: one row per manifest after the rewrite
SELECT path,
       added_data_files_count + existing_data_files_count AS files_tracked,
       partition_summaries
FROM lake.gis.telemetry.manifests
ORDER BY files_tracked DESC;

Finally, inspect that a new snapshot with a replace operation was recorded — a sort-based rewrite_data_files commits as a replace, never an append or overwrite, which is your proof the operation was layout-only:

SELECT snapshot_id, operation, summary['total-data-files'] AS files
FROM lake.gis.telemetry.snapshots
ORDER BY committed_at DESC LIMIT 3;

If operation reads anything other than replace, the rewrite did more than reorganize files and you should investigate before trusting the result. You can also compare summary['total-data-files'] between the two most recent snapshots to quantify the reduction — going from, say, 512 to 47 files is the number to log and alert on.

The same clustering principle underpins Z-ordering; for the multi-dimensional variant used on spatial joins see Z-ordering for geospatial queries, and for the Delta-side equivalent of this whole flow see scheduling VACUUM for spatial Delta tables. The authoritative parameter reference is the Iceberg Spark procedures documentation and the maintenance guide.

Choosing a Rewrite Strategy

rewrite_data_files offers more than one strategy, and the choice changes both the cost and the benefit substantially on a spatial table.

The pattern that balances cost against benefit is to run binpack frequently against the open partition — it is cheap, needs no shuffle, and keeps the file count under control as micro-batches arrive — and to run the Z-order rewrite once, when the partition closes and will not be written to again. The expensive operation then happens exactly once per partition for the lifetime of the table.

Scoping the Rewrite

The middle limit is the one most often left at its default and most worth tuning on a spatial table, because geometry-heavy files reach the target size at far lower row counts than scalar data. Setting it from the observed size distribution rather than from a general default avoids rewriting files that were already fine. Measure the distribution once with a metadata query, set the threshold from it, and revisit only when the data shape changes.