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.