If you take care of massive analytical datasets saved in desk codecs akin to Apache Iceberg, there’s a identified situation known as the small recordsdata downside. This occurs when information is written in a lot of tiny recordsdata fairly than a smaller variety of fairly sized ones. This will increase metadata overhead and may decelerate question planning and execution.
To fight this, a course of known as compaction is used which mixes all of the smaller recordsdata right into a small variety of bigger recordsdata. This reduces metadata overhead and provides the question engine fewer recordsdata to open, scan and handle, which may enhance efficiency considerably.
I’m focussing on Apache Iceberg because it’s quickly rising into one of many main open desk codecs for large-scale analytical information, bringing options akin to schema evolution, time journey, partition evolution and dependable transactions to information saved in object storage or distributed file programs.
Importantly, though Iceberg helps compaction it doesn’t do it for us mechanically. Iceberg provides the rewrite_data_files process and the metadata wanted to pick and course of recordsdata for compaction however we because the system admins nonetheless must execute that process or configure one other system to set off it.
However, actually, does compaction make that a lot of a distinction? That’s the query this text will attempt to reply. We’ll create an Iceberg desk containing 50 million rows unfold throughout 1,000 tiny recordsdata. These recordsdata will stay untouched till we situation the compaction command. We’ll measure three SQL workloads earlier than and after the rewrite. That may inform us if compaction is value it.
All the pieces runs domestically. You received’t want a cloud account, Docker, a Hadoop cluster, or a paid service. The draw back of this setup is that we will’t reproduce really huge datasets which are utilized in real-world programs however hopefully our outcomes will give us some helpful insights.
What’s Apache Iceberg?
Apache Iceberg is an open desk format for accessing big analytic datasets, bringing database-like options akin to schema evolution, partition evolution, time journey, and dependable transactions to information saved in recordsdata akin to Parquet. Parquet is a file format. It determines how rows and columns are encoded inside a person file. Iceberg operates one degree above that.
An Iceberg desk usually incorporates Parquet, Avro or ORC information recordsdata, plus metadata describing which recordsdata presently belong to the desk. Its snapshots present a historical past of desk modifications, whereas manifest recordsdata assist question engines find related information recordsdata with out recursively itemizing each listing.
This further metadata lets engines deal with a set of recordsdata extra like a database desk. Iceberg helps atomic modifications, schema evolution, partition evolution and time-travel queries with out changing the underlying information right into a proprietary storage format.
It may’t, nonetheless, stop each poor write sample.
Suppose a streaming job writes a small batch each minute. Every batch could produce a number of new recordsdata. After a month, a modest amount of information will be scattered throughout tens of 1000’s of objects. A question engine should plan work for these recordsdata, open them, learn their metadata and shut them once more.
The identical factor can occur with batch processing. Spark writes output from its duties independently, and a file can’t span an Iceberg partition boundary. Iceberg’s write.target-file-size-bytes property is subsequently a “finest endeavours” operation, not an absolute promise. The Iceberg documentation explicitly notes that Spark can’t write a file bigger than the duty producing it. A 512 MB goal is irrelevant if a job solely has sufficient information to create a 70 KB Parquet file.
Compaction fixes this by studying small recordsdata and rewriting their rows into fewer, bigger recordsdata. This usually means much less work is rquired to learn and course of these recordsdata. For this demo, we’re going to check Iceberg’s default bin-pack technique, which modifications the packaging with out intentionally sorting the rows.
The Apache Iceberg documentation classifies data-file compaction as elective upkeep. It tells us to examine the recordsdata metadata desk and run rewriteDataFiles when acceptable. The format offers the operation, however deciding when to run it stays a part of working the desk.
Establishing a dev surroundings and putting in the required software program
Earlier than writing any code, let’s arrange a growth surroundings to maintain the venture remoted. I exploit the uv software for this, however use whichever technique you recognize finest.
REM Create a brand new venture folder and swap to itc:> mkdir C:iceberg-compactionc:> cd /d C:iceberg-compactionREM Replace uv softwareC:iceberg-compaction> uv self replacedata: Checking for updates…success: You are already on model v0.12.5 of uv (the most recent model).
That is what we’ll want for our experiment. If you have already got some or all of those, go away effectively alone and set up simply those you want.
Python 3.10–3.13
Java 17 or Java 21
PySpark 4.0.3
Apache Iceberg 1.11.0
REM Set up JAVAc:iceberg-compaction> winget set up EclipseAdoptium.Temurin.21.JDKDiscovered Eclipse Temurin JDK with Hotspot 21 [EclipseAdoptium.Temurin.21.JDK] Model 21.0.12.101This software is licensed to you by its proprietor.Microsoft will not be liable for, nor does it grant any licenses to, third-party packages.Downloading https://github.com/adoptium/temurin21-binaries/releases/obtain/jdk-21.0.12.1+1/OpenJDK21U-jdk_x64_windows_hotspot_21.0.12.1_1.msi ██████████████████████████████ 171 MB / 171 MBEfficiently verified installer hashBeginning bundle set up...Efficiently put inC:iceberg-compaction> java -versionopenjdk model "21.0.12.1" 2026-08-18 LTSOpenJDK Runtime Surroundings Temurin-21.0.12.1+1 (construct 21.0.12.1+1-LTS)OpenJDK 64-Bit Server VM Temurin-21.0.12.1+1 (construct 21.0.12.1+1-LTS, blended mode, sharing)REM Create and swap to a brand new surroundings with Python 3.13C:iceberg-compaction> uv initInitialized venture `iceberg-compaction`C:iceberg-compaction> uv venv --python 3.13C:iceberg-compaction> .venvScriptsactivate(iceberg-compaction) C:iceberg-compaction> python --versionPython 3.13.1REM Set up Spark(iceberg-compaction) C:iceberg-compaction> uv add pyspark==4.0.3Resolved 3 packages in 42.96s Constructed pyspark==4.0.3 Constructed iceberg-compaction @ file:///C:/Customers/thoma/initiatives/iceberg-compaction Ready 3 packages in 33.86s░░░░░░░░░░░░░░░░░░░░ [0/3] Putting in wheels... warning: Did not hardlink recordsdata; falling again to full copy. This may increasingly result in degraded efficiency. If the cache and goal directories are on completely different filesystems, hardlinking is probably not supported. If that is intentional, set `export UV_LINK_MODE=copy` or use `--link-mode=copy` to suppress this warning.Put in 3 packages in 2.28s + iceberg-compaction==0.1.0 (from file:///C:/Customers/thoma/initiatives/iceberg-compaction) + py4j==0.10.9.9 + pyspark==4.0.3
Iceberg has no separate Python set up step right 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 domestically. You’ll see that occur once we run the script later.
Writing the Python code
Create a file named iceberg_compaction_demo.py. The code blocks under type one script and must be added within the order proven.
The 64 MiB goal is massive sufficient to make compaction significant on a laptop computer with out turning our native take a look at into an all-day job. Iceberg’s regular data-file goal is 512 MB.
Creating the Spark session
Subsequent, create the Spark session (you possibly can see the reference to Iceberg 1.11.0 that I talked about earlier):
native[*] tells Spark to make use of the out there logical processors. Our catalogue is known as native, makes use of Iceberg’s HadoopCatalog, and shops every little thing beneath the iceberg_lab_warehouse listing.
Regardless of its identify, this catalogue doesn’t require Hadoop to be operating. It makes use of Hadoop’s filesystem interface to handle an extraordinary native listing. A Hadoop catalogue on an area filesystem isn’t protected for concurrent writers, however that limitation is OK for this single-process experiment.
Adaptive Question Execution is disabled as a result of Spark would possibly in any other case mix our intentionally small duties. That may be smart behaviour in an actual workload however would spoil the demonstration.
On its first run, Spark downloads the roughly 46 MB Iceberg runtime from Maven Central. It caches the JAR, so later runs don’t usually obtain it once more.
Producing deterministic take a look at information
This information is intentionally non-random, so every run creates the identical rows and question outcome. The ultimate repartition forces the DataFrame by means of the requested variety of Spark duties. As a result of every job writes independently, asking for 1,000 duties provides us 1,000 small information recordsdata. Setting 64 shuffle partitions retains the aggregation benchmarks from creating 1,000 result-side duties.
Creating the desk
if WAREHOUSE.exists(): shutil.rmtree(WAREHOUSE)spark = build_spark()spark.sparkContext.setLogLevel("WARN")spark.sql("CREATE NAMESPACE IF NOT EXISTS native.lab")spark.sql( f""" CREATE TABLE {TABLE} ( id BIGINT, customer_id INT, event_type STRING, event_date DATE, quantity 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 first of each run. Don’t level the WAREHOUSE surroundings variable at a listing containing something that you must preserve!
The distribution mode is disabled so Iceberg doesn’t reorganise our rigorously fragmented enter earlier than writing it.
Asking Iceberg about its recordsdata
Counting recordsdata within the listing is unreliable as a result of Iceberg retains outdated recordsdata for historic snapshots. As an alternative, question the desk’s recordsdata metadata desk, which describes recordsdata belonging to the present snapshot:
def file_statistics(spark: SparkSession, label: str) -> dict: print(f"n{label}") outcome = spark.sql( f""" SELECT COUNT(*) AS data_files, SUM(record_count) AS data, 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}.recordsdata WHERE content material = 0 """ ) row = outcome.first() print( f"{int(row['data_files']):,} lively recordsdata, " f"{int(row['records']):,} data, " f"{float(row['total_mib']):,.2f} MiB" ) return row.asDict()
As anticipated, my run produced 1,000 information recordsdata containing 50 million data.
Establishing our SQL benchmarks
One question would inform us little or no, so the take a look at suite makes use of three workloads:
A filtered aggregation for a variety of shoppers
A full-table aggregation grouped by occasion date
A slim lookup protecting 10,000 consecutive IDs
Add these queries and benchmark operate:
QUERIES = { "Filtered buyer aggregation": f""" SELECT event_type, COUNT(*) AS occasions, ROUND(SUM(CAST(quantity 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 each day aggregation": f""" SELECT event_date, COUNT(*) AS occasions, ROUND(AVG(CAST(quantity AS DOUBLE)), 2) AS average_amount FROM {TABLE} GROUP BY event_date ORDER BY event_date """, "Slim ID-range lookup": f""" SELECT COUNT(*) AS occasions, ROUND(SUM(CAST(quantity 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 identify, question in QUERIES.gadgets(): # One unreported run warms the JVM and reads the desk metadata. anticipated = spark.sql(question).accumulate() timings = [] for _ in vary(repetitions): spark.catalog.clearCache() began = time.perf_counter() precise = spark.sql(question).accumulate() timings.append(time.perf_counter() - began) if precise != anticipated: elevate RuntimeError(f"{identify} returned inconsistent outcomes") median = statistics.median(timings) measurements[name] = {"rows": anticipated, "median": median} print(f"n{identify} ({len(anticipated)} outcome rows)") print("Instances (seconds):", ", ".be part of(f"{worth:.3f}" for worth in timings)) print(f"Median: {median:.3f} seconds") return measurements
The unreported first execution of every question lets the JVM initialise and masses Iceberg’s metadata. 5 measured executions are extra informative than deciding on whichever single run helps 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']):,}", ), ( "Data", f"{int(before_files['records']):,}", f"{int(after_files['records']):,}", ), ( "Whole lively information measurement", f"{float(before_files['total_mib']):,.2f} MiB", f"{float(after_files['total_mib']):,.2f} MiB", ), ( "Common file measurement", format_average_file_size(before_files["average_kib"]), format_average_file_size(after_files["average_kib"]), ), ] for identify in QUERIES: rows.append( ( identify, f"{before_queries[name]['median']:.3f} s", f"{after_queries[name]['median']:.3f} s", ) ) headers = ("Measurement", "Earlier than", "After") widths = [ max(len(headers[index]), *(len(row[index]) for row in rows)) for index in vary(3) ] border = "+" + "+".be part of("-" * (width + 2) for width in widths) + "+" def print_row(row): cells = [row[index].ljust(widths[index]) for index in vary(3)] print("| " + " | ".be part of(cells) + " |") print("nBefore/after comparability") print(border) print_row(headers) print(border) for row in rows: print_row(row) print(border)
Fundamental driver code
def foremost() -> None: print(f"PySpark goal model: {SPARK_VERSION}") print(f"Iceberg model: {ICEBERG_VERSION}") print(f"Warehouse: {WAREHOUSE}") print("The warehouse listing is deleted and recreated on each run.") if WAREHOUSE.exists(): shutil.rmtree(WAREHOUSE) spark = build_spark() spark.sparkContext.setLogLevel("WARN") strive: print(f"Working Spark {spark.model}") if spark.model != SPARK_VERSION: print( f"WARNING: this experiment was written for Spark {SPARK_VERSION}, " f"however {spark.model} is operating." ) spark.sql("CREATE NAMESPACE IF NOT EXISTS native.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, quantity DECIMAL(10, 2) ) USING iceberg TBLPROPERTIES ( 'write.distribution-mode' = 'none', 'write.target-file-size-bytes' = '{TARGET_FILE_SIZE}' ) """ ) print(f"nWriting {ROWS:,} rows by means of {INITIAL_FILES} Spark duties...") make_events(spark, 0, ROWS, INITIAL_FILES).writeTo(TABLE).append() before_files = file_statistics(spark, "Earlier than compaction") earlier than = benchmark(spark, "Earlier than compaction") print("nCompacting the desk with Iceberg's bin-pack technique...") compaction = spark.sql( f""" CALL native.system.rewrite_data_files( desk => 'native.lab.occasions', technique => 'binpack', choices => map( 'target-file-size-bytes', '{TARGET_FILE_SIZE}', 'min-input-files', '2' ) ) """ ) compaction.present(truncate=False) after_files = file_statistics(spark, "After compaction") after = benchmark(spark, "After compaction") for identify in QUERIES: if earlier than[name]["rows"] != after[name]["rows"]: elevate RuntimeError(f"{identify} modified after compaction") print_comparison(before_files, after_files, earlier than, after) print( "nFinished. The present desk is unbroken in iceberg_lab_warehouse. " "Previous recordsdata are additionally retained as a result of Iceberg snapshots nonetheless refer " "to them." ) lastly: spark.cease()if __name__ == "__main__": foremost()
The outcome examine on this part is necessary.
......for identify in QUERIES: if earlier than[name]["rows"] != after[name]["rows"]: elevate RuntimeError(f"{identify} modified after compaction")......
Compaction should change the desk’s bodily recordsdata with out altering a row returned by any question.
Working the demo
(iceberg-compaction) C:iceberg-compaction> python iceberg_compaction_demo.pyPySpark goal model: 4.0.3Iceberg model: 1.11.0Warehouse: D:iceberg-compactioniceberg_lab_warehouseThe warehouse listing is deleted and recreated on each run.WARNING: Utilizing incubator modules: jdk.incubator.vector......Writing 50,000,000 rows by means of 1000 Spark duties...Earlier than compaction1,000 lively recordsdata, 50,000,000 data, 380.29 MiB......After compaction6 lively recordsdata, 50,000,000 data, 357.54 MiBAfter compactionFiltered buyer aggregation (4 outcome rows)Instances (seconds): 0.435, 0.448, 0.446, 0.574, 0.431Median: 0.446 secondsFull-table each day aggregation (31 outcome rows)Instances (seconds): 0.582, 0.573, 0.644, 0.567, 0.563Median: 0.573 secondsSlim ID-range lookup (1 outcome rows)Instances (seconds): 0.964, 0.402, 0.319, 0.300, 0.387Median: 0.387 secondsEarlier than/after comparability+-------------------------------+------------+------------+| Measurement | Earlier than | After |+-------------------------------+------------+------------+| Lively information recordsdata | 1,000 | 6 || Data | 50,000,000 | 50,000,000 || Whole lively information measurement | 380.29 MiB | 357.54 MiB || Common file measurement | 389.42 KiB | 59.59 MiB || Filtered buyer aggregation | 1.013 s | 0.446 s || Full-table each day aggregation | 1.580 s | 0.573 s || Slim ID-range lookup | 0.564 s | 0.387 s |+-------------------------------+------------+------------+
On my fairly well-specc’ed desktop, all three median instances improved between 31% and 63% after compaction. Not too shabby!
Your numbers will differ. We’re testing native file format, CPU scheduling and filesystem caching in addition to Iceberg. The helpful comparability is earlier than versus after on the identical machine.
This isn’t proof that compaction makes each question 63% quicker. The compressed desk continues to be solely 380 MiB, and Spark’s fastened job overhead accounts for a few of its runtime. The experiment establishes the mechanism: the question engine has fewer recordsdata to plan and open. The profit in an actual desk relies on its storage, file depend, filters, partitions, engine and workload.
Compaction additionally carried out a whole learn and rewrite of the desk. That work isn’t free. A desk queried as soon as could by no means get well the price of compacting it.
Abstract
Compaction is a vital a part of working Iceberg tables that accumulate massive numbers of small recordsdata. When small-file fragmentation turns into important, compaction can considerably enhance question efficiency – however whether or not and the way typically it ought to run relies on the workload.
Keep in mind that as quickly as you begin to write extra information and recordsdata, the advantages of compaction shall be misplaced over time. For this reason compaction requires a coverage.
You’ll in all probability must spend some effort and time guaranteeing you’re compacting on the proper frequency to your particular workloads.
A smart coverage relies upon solely on your corporation wants. You can schedule it each night time throughout a quiet interval or after a sure variety of recordsdata are created. Iceberg’s rewrite_data_files process accepts a the place parameter, so upkeep can goal latest or significantly fragmented partitions as a substitute of rewriting a complete desk.
You additionally want to contemplate file measurement. Bigger recordsdata scale back file-opening and metadata overhead, however additionally they scale back learn parallelism and make every rewrite extra substantial.
One last consideration that always causes confusion: these “outdated” recordsdata you simply compacted are nonetheless there. That’s as a result of Iceberg wants them for doing time-travel queries. Compaction modifications the present snapshot; it isn’t snapshot expiration or orphan-file elimination. These are separate upkeep operations with separate retention selections.
The small-file downside subsequently has no one-off repair. Iceberg offers the compaction operations, however it may’t know the way typically a desk is queried, how a lot upkeep capability is accessible or how lengthy historic snapshots should survive.
Compaction turns many small recordsdata into fewer massive ones. Knowledge engineering begins with deciding when that rewrite is value doing.
