When you deal with large analytical datasets stored in table formats such as Apache Iceberg, there is a known issue called the small files problem. This happens when data is written in lots of tiny files rather than a smaller number of reasonably sized ones. This increases metadata overhead and can slow down query planning and execution.
To combat this, a process called compaction is used which combines all the smaller files into a small number of larger files. This reduces metadata overhead and gives the query engine fewer files to open, scan and manage, which can improve performance significantly.
I’m focussing on Apache Iceberg as it’s rapidly growing into one of the leading open table formats for large-scale analytical data, bringing features such as schema evolution, time travel, partition evolution and reliable transactions to data stored in object storage or distributed file systems.
Importantly, although Iceberg supports compaction it doesn’t do it for us automatically. Iceberg supplies the rewrite_data_files procedure and the metadata needed to select and process files for compaction but we as the system admins still have to execute that procedure or configure another system to trigger it.
But, really, does compaction make that much of a difference? That’s the question this article will try to answer. We’ll create an Iceberg table containing 50 million rows spread across 1,000 tiny files. Those files will remain untouched until we issue the compaction command. We’ll measure three SQL workloads before and after the rewrite. That will tell us if compaction is worth it.
Everything runs locally. You won’t need a cloud account, Docker, a Hadoop cluster, or a paid service. The downside of this setup is that we can’t reproduce truly massive datasets that are used in real-world systems but hopefully our results will give us some useful insights.
What is Apache Iceberg?
Apache Iceberg is an open table format for accessing huge analytic datasets, bringing database-like features such as schema evolution, partition evolution, time travel, and reliable transactions to data stored in files such as Parquet. Parquet is a file format. It determines how rows and columns are encoded inside an individual file. Iceberg operates one level above that.
An Iceberg table normally contains Parquet, Avro or ORC data files, plus metadata describing which files currently belong to the table. Its snapshots provide a history of table changes, while manifest files help query engines locate relevant data files without recursively listing every directory.
This extra metadata lets engines treat a collection of files more like a database table. Iceberg supports atomic changes, schema evolution, partition evolution and time-travel queries without converting the underlying data into a proprietary storage format.
It can’t, however, prevent every poor write pattern.
Suppose a streaming job writes a small batch every minute. Each batch may produce one or more new files. After a month, a modest quantity of data can be scattered across tens of thousands of objects. A query engine must plan work for those files, open them, read their metadata and close them again.
The same thing can happen with batch processing. Spark writes output from its tasks independently, and a file can’t span an Iceberg partition boundary. Iceberg’s write.target-file-size-bytes property is therefore a “best endeavours” operation, not an absolute promise. The Iceberg documentation explicitly notes that Spark can’t write a file larger than the task producing it. A 512 MB target is irrelevant if a task only has enough data to create a 70 KB Parquet file.
Compaction fixes this by reading small files and rewriting their rows into fewer, larger files. This generally means less work is rquired to read and process those files. For this demo, we’re going to test Iceberg’s default bin-pack strategy, which changes the packaging without deliberately sorting the rows.
The Apache Iceberg documentation classifies data-file compaction as optional maintenance. It tells us to inspect the files metadata table and run rewriteDataFiles when appropriate. The format provides the operation, but deciding when to run it remains part of operating the table.
Setting up a dev environment and installing the required software
Before writing any code, let’s set up a development environment to keep the project isolated. I use the uv tool for this, but use whichever method you know best.
REM Create a new project folder and switch to itc:> mkdir C:iceberg-compactionc:> cd /d C:iceberg-compactionREM Update uv toolC:iceberg-compaction> uv self updateinfo: Checking for updates…success: You're already on version v0.12.5 of uv (the latest version).
This is what we’ll need for our experiment. If you already have some or all of these, leave well alone and install just the ones you need.
Python 3.10–3.13
Java 17 or Java 21
PySpark 4.0.3
Apache Iceberg 1.11.0
REM Install JAVAc:iceberg-compaction> winget install EclipseAdoptium.Temurin.21.JDKFound Eclipse Temurin JDK with Hotspot 21 [EclipseAdoptium.Temurin.21.JDK] Version 21.0.12.101This application is licensed to you by its owner.Microsoft is not responsible for, nor does it grant any licenses to, third-party packages.Downloading https://github.com/adoptium/temurin21-binaries/releases/download/jdk-21.0.12.1+1/OpenJDK21U-jdk_x64_windows_hotspot_21.0.12.1_1.msi ██████████████████████████████ 171 MB / 171 MBSuccessfully verified installer hashStarting package install...Successfully installedC:iceberg-compaction> java -versionopenjdk version "21.0.12.1" 2026-08-18 LTSOpenJDK Runtime Environment Temurin-21.0.12.1+1 (build 21.0.12.1+1-LTS)OpenJDK 64-Bit Server VM Temurin-21.0.12.1+1 (build 21.0.12.1+1-LTS, mixed mode, sharing)REM Create and switch to a new environment with Python 3.13C:iceberg-compaction> uv initInitialized project `iceberg-compaction`C:iceberg-compaction> uv venv --python 3.13C:iceberg-compaction> .venvScriptsactivate(iceberg-compaction) C:iceberg-compaction> python --versionPython 3.13.1REM Install Spark(iceberg-compaction) C:iceberg-compaction> uv add pyspark==4.0.3Resolved 3 packages in 42.96s Built pyspark==4.0.3 Built iceberg-compaction @ file:///C:/Users/thoma/projects/iceberg-compaction Prepared 3 packages in 33.86s░░░░░░░░░░░░░░░░░░░░ [0/3] Installing wheels... warning: Failed to hardlink files; falling back to full copy. This may lead to degraded performance. If the cache and target directories are on different filesystems, hardlinking may not be supported. If this is intentional, set `export UV_LINK_MODE=copy` or use `--link-mode=copy` to suppress this warning.Installed 3 packages in 2.28s + iceberg-compaction==0.1.0 (from file:///C:/Users/thoma/projects/iceberg-compaction) + py4j==0.10.9.9 + pyspark==4.0.3
Iceberg has no separate Python installation step here. When the Spark session is created, Spark resolves the iceberg-spark-runtime-4.0_2.13:1.11.0 dependency from Maven Central and caches the JAR locally. You’ll see that happen when we run the script later.
Writing the Python code
Create a file named iceberg_compaction_demo.py. The code blocks below form one script and should be added in the order shown.
Configuring a local Iceberg catalogue
from __future__ import annotationsimport osimport shutilimport statisticsimport timefrom pathlib import Pathfrom pyspark.sql import SparkSession, functions as FSPARK_VERSION = "4.0.3"ICEBERG_VERSION = "1.11.0"ROWS = 50_000_000INITIAL_FILES = 1_000TARGET_FILE_SIZE = 64 * 1024 * 1024 # 64 MiB: suitable for this local demoWAREHOUSE = (Path.cwd() / "iceberg_lab_warehouse").resolve()TABLE = "local.lab.events"
The 64 MiB target is large enough to make compaction meaningful on a laptop without turning our local test into an all-day job. Iceberg’s normal data-file target is 512 MB.
Creating the Spark session
Next, create the Spark session (you can see the reference to Iceberg 1.11.0 that I talked about earlier):
local[*] tells Spark to use the available logical processors. Our catalogue is named local, uses Iceberg’s HadoopCatalog, and stores everything beneath the iceberg_lab_warehouse directory.
Despite its name, this catalogue doesn’t require Hadoop to be running. It uses Hadoop’s filesystem interface to manage an ordinary local directory. A Hadoop catalogue on a local filesystem isn’t safe for concurrent writers, but that limitation is OK for this single-process experiment.
Adaptive Query Execution is disabled because Spark might otherwise combine our deliberately small tasks. That would be sensible behaviour in a real workload but would spoil the demonstration.
On its first run, Spark downloads the approximately 46 MB Iceberg runtime from Maven Central. It caches the JAR, so later runs don’t normally download it again.
This data is deliberately non-random, so each run creates the same rows and query result. The final repartition forces the DataFrame through the requested number of Spark tasks. Because each task writes independently, asking for 1,000 tasks gives us 1,000 small data files. Setting 64 shuffle partitions keeps the aggregation benchmarks from creating 1,000 result-side tasks.
Creating the table
if WAREHOUSE.exists(): shutil.rmtree(WAREHOUSE)spark = build_spark()spark.sparkContext.setLogLevel("WARN")spark.sql("CREATE NAMESPACE IF NOT EXISTS local.lab")spark.sql( f""" CREATE TABLE {TABLE} ( id BIGINT, customer_id INT, event_type STRING, event_date DATE, amount DECIMAL(10, 2) ) USING iceberg TBLPROPERTIES ( 'write.distribution-mode' = 'none', 'write.target-file-size-bytes' = '{TARGET_FILE_SIZE}' ) """)make_events(spark, 0, ROWS, INITIAL_FILES).writeTo(TABLE).append()
For repeatability, the code deletes the iceberg_lab_warehouse folder at the beginning of every run. Don’t point the WAREHOUSE environment variable at a directory containing anything you need to keep!
The distribution mode is disabled so Iceberg doesn’t reorganise our carefully fragmented input before writing it.
Asking Iceberg about its files
Counting files in the directory is unreliable because Iceberg retains old files for historical snapshots. Instead, query the table’s files metadata table, which describes files belonging to the current snapshot:
def file_statistics(spark: SparkSession, label: str) -> dict: print(f"n{label}") result = spark.sql( f""" SELECT COUNT(*) AS data_files, SUM(record_count) AS records, ROUND(SUM(file_size_in_bytes) / 1048576.0, 2) AS total_mib, ROUND(AVG(file_size_in_bytes) / 1024.0, 2) AS average_kib, ROUND(MIN(file_size_in_bytes) / 1024.0, 2) AS smallest_kib, ROUND(MAX(file_size_in_bytes) / 1024.0, 2) AS largest_kib FROM {TABLE}.files WHERE content = 0 """ ) row = result.first() print( f"{int(row['data_files']):,} active files, " f"{int(row['records']):,} records, " f"{float(row['total_mib']):,.2f} MiB" ) return row.asDict()
As expected, my run produced 1,000 data files containing 50 million records.
Establishing our SQL benchmarks
One query would tell us very little, so the test suite uses three workloads:
A filtered aggregation for a range of customers
A full-table aggregation grouped by event date
A narrow lookup covering 10,000 consecutive IDs
Add these queries and benchmark function:
QUERIES = { "Filtered customer aggregation": f""" SELECT event_type, COUNT(*) AS events, ROUND(SUM(CAST(amount AS DOUBLE)), 2) AS total_amount FROM {TABLE} WHERE customer_id BETWEEN 1000 AND 1999 GROUP BY event_type ORDER BY event_type """, "Full-table daily aggregation": f""" SELECT event_date, COUNT(*) AS events, ROUND(AVG(CAST(amount AS DOUBLE)), 2) AS average_amount FROM {TABLE} GROUP BY event_date ORDER BY event_date """, "Narrow ID-range lookup": f""" SELECT COUNT(*) AS events, ROUND(SUM(CAST(amount AS DOUBLE)), 2) AS total_amount FROM {TABLE} WHERE id BETWEEN 500000 AND 509999 """,}def benchmark(spark: SparkSession, label: str, repetitions: int = 5): print(f"n{label}") measurements = {} for name, query in QUERIES.items(): # One unreported run warms the JVM and reads the table metadata. expected = spark.sql(query).collect() timings = [] for _ in range(repetitions): spark.catalog.clearCache() started = time.perf_counter() actual = spark.sql(query).collect() timings.append(time.perf_counter() - started) if actual != expected: raise RuntimeError(f"{name} returned inconsistent results") median = statistics.median(timings) measurements[name] = {"rows": expected, "median": median} print(f"n{name} ({len(expected)} result rows)") print("Times (seconds):", ", ".join(f"{value:.3f}" for value in timings)) print(f"Median: {median:.3f} seconds") return measurements
The unreported first execution of each query lets the JVM initialise and loads Iceberg’s metadata. Five measured executions are more informative than selecting whichever single run supports the argument.
Prettify the output
def format_average_file_size(value_kib) -> str: value_kib = float(value_kib) if value_kib >= 1024: return f"{value_kib / 1024:,.2f} MiB" return f"{value_kib:,.2f} KiB"def print_comparison(before_files, after_files, before_queries, after_queries): rows = [ ( "Active data files", f"{int(before_files['data_files']):,}", f"{int(after_files['data_files']):,}", ), ( "Records", f"{int(before_files['records']):,}", f"{int(after_files['records']):,}", ), ( "Total active data size", f"{float(before_files['total_mib']):,.2f} MiB", f"{float(after_files['total_mib']):,.2f} MiB", ), ( "Average file size", format_average_file_size(before_files["average_kib"]), format_average_file_size(after_files["average_kib"]), ), ] for name in QUERIES: rows.append( ( name, f"{before_queries[name]['median']:.3f} s", f"{after_queries[name]['median']:.3f} s", ) ) headers = ("Measurement", "Before", "After") widths = [ max(len(headers[index]), *(len(row[index]) for row in rows)) for index in range(3) ] border = "+" + "+".join("-" * (width + 2) for width in widths) + "+" def print_row(row): cells = [row[index].ljust(widths[index]) for index in range(3)] print("| " + " | ".join(cells) + " |") print("nBefore/after comparison") print(border) print_row(headers) print(border) for row in rows: print_row(row) print(border)
Main driver code
def main() -> None: print(f"PySpark target version: {SPARK_VERSION}") print(f"Iceberg version: {ICEBERG_VERSION}") print(f"Warehouse: {WAREHOUSE}") print("The warehouse directory is deleted and recreated on every run.") if WAREHOUSE.exists(): shutil.rmtree(WAREHOUSE) spark = build_spark() spark.sparkContext.setLogLevel("WARN") try: print(f"Running Spark {spark.version}") if spark.version != SPARK_VERSION: print( f"WARNING: this experiment was written for Spark {SPARK_VERSION}, " f"but {spark.version} is running." ) spark.sql("CREATE NAMESPACE IF NOT EXISTS local.lab") spark.sql(f"DROP TABLE IF EXISTS {TABLE}") spark.sql( f""" CREATE TABLE {TABLE} ( id BIGINT, customer_id INT, event_type STRING, event_date DATE, amount DECIMAL(10, 2) ) USING iceberg TBLPROPERTIES ( 'write.distribution-mode' = 'none', 'write.target-file-size-bytes' = '{TARGET_FILE_SIZE}' ) """ ) print(f"nWriting {ROWS:,} rows through {INITIAL_FILES} Spark tasks...") make_events(spark, 0, ROWS, INITIAL_FILES).writeTo(TABLE).append() before_files = file_statistics(spark, "Before compaction") before = benchmark(spark, "Before compaction") print("nCompacting the table with Iceberg's bin-pack strategy...") compaction = spark.sql( f""" CALL local.system.rewrite_data_files( table => 'local.lab.events', strategy => 'binpack', options => map( 'target-file-size-bytes', '{TARGET_FILE_SIZE}', 'min-input-files', '2' ) ) """ ) compaction.show(truncate=False) after_files = file_statistics(spark, "After compaction") after = benchmark(spark, "After compaction") for name in QUERIES: if before[name]["rows"] != after[name]["rows"]: raise RuntimeError(f"{name} changed after compaction") print_comparison(before_files, after_files, before, after) print( "nFinished. The current table is intact in iceberg_lab_warehouse. " "Old files are also retained because Iceberg snapshots still refer " "to them." ) finally: spark.stop()if __name__ == "__main__": main()
The result check in this section is important.
......for name in QUERIES: if before[name]["rows"] != after[name]["rows"]: raise RuntimeError(f"{name} changed after compaction")......
Compaction must change the table’s physical files without changing a row returned by any query.
Running the demo
(iceberg-compaction) C:iceberg-compaction> python iceberg_compaction_demo.pyPySpark target version: 4.0.3Iceberg version: 1.11.0Warehouse: D:iceberg-compactioniceberg_lab_warehouseThe warehouse directory is deleted and recreated on every run.WARNING: Using incubator modules: jdk.incubator.vector......Writing 50,000,000 rows through 1000 Spark tasks...Before compaction1,000 active files, 50,000,000 records, 380.29 MiB......After compaction6 active files, 50,000,000 records, 357.54 MiBAfter compactionFiltered customer aggregation (4 result rows)Times (seconds): 0.435, 0.448, 0.446, 0.574, 0.431Median: 0.446 secondsFull-table daily aggregation (31 result rows)Times (seconds): 0.582, 0.573, 0.644, 0.567, 0.563Median: 0.573 secondsNarrow ID-range lookup (1 result rows)Times (seconds): 0.964, 0.402, 0.319, 0.300, 0.387Median: 0.387 secondsBefore/after comparison+-------------------------------+------------+------------+| Measurement | Before | After |+-------------------------------+------------+------------+| Active data files | 1,000 | 6 || Records | 50,000,000 | 50,000,000 || Total active data size | 380.29 MiB | 357.54 MiB || Average file size | 389.42 KiB | 59.59 MiB || Filtered customer aggregation | 1.013 s | 0.446 s || Full-table daily aggregation | 1.580 s | 0.573 s || Narrow ID-range lookup | 0.564 s | 0.387 s |+-------------------------------+------------+------------+
On my reasonably well-specc’ed desktop, all three median times improved between 31% and 63% after compaction. Not too shabby!
Your numbers will differ. We’re testing local file layout, CPU scheduling and filesystem caching as well as Iceberg. The useful comparison is before versus after on the same machine.
This isn’t evidence that compaction makes every query 63% faster. The compressed table is still only 380 MiB, and Spark’s fixed job overhead accounts for some of its runtime. The experiment establishes the mechanism: the query engine has fewer files to plan and open. The benefit in a real table depends on its storage, file count, filters, partitions, engine and workload.
Compaction also performed a complete read and rewrite of the table. That work isn’t free. A table queried once may never recover the cost of compacting it.
Summary
Compaction is an important part of operating Iceberg tables that accumulate large numbers of small files. When small-file fragmentation becomes significant, compaction can substantially improve query performance – but whether and how often it should run depends on the workload.
Bear in mind that as soon as you start to write more data and files, the benefits of compaction will be lost over time. This is why compaction requires a policy.
You’ll probably need to spend some time and effort ensuring you’re compacting at the right frequency for your specific workloads.
A sensible policy depends entirely on your business needs. You could schedule it every night during a quiet period or after a certain number of files are created. Iceberg’s rewrite_data_files procedure accepts a where parameter, so maintenance can target recent or particularly fragmented partitions instead of rewriting an entire table.
You also need to consider file size. Larger files reduce file-opening and metadata overhead, but they also reduce read parallelism and make each rewrite more substantial.
One final consideration that often causes confusion: those “old” files you just compacted are still there. That’s because Iceberg needs them for doing time-travel queries. Compaction changes the current snapshot; it isn’t snapshot expiration or orphan-file removal. Those are separate maintenance operations with separate retention decisions.
The small-file problem therefore has no one-off fix. Iceberg provides the compaction operations, but it can’t know how often a table is queried, how much maintenance capacity is available or how long historical snapshots must survive.
Compaction turns many small files into fewer large ones. Data engineering begins with deciding when that rewrite is worth doing.
