How to efficiently load large data into Iceberg?

I've actually worked with initializing 10TB iceberg tables (with around 30k partitions). In the project we have another datasource (basically around 25k data-files in another format) and we have a spark job that says "every original data-file is a conversion-task, all tasks belong to the same stage, the result of the conversion-tasks is added to the table". As the documentation says, the final step should look like this: df.writeTo("prod.db.table").append() As for problems: By default, doing a huge write-job will have a huge shuffle step, so rather than writing 10TB of parquet-files, we found ourselves doing 10TB of shuffle-write (too large for our cluster). This makes sure that even if each of the 25k tasks produces rows belonging to the 10k different partitions, we won't end up with 250M tiny files (it does an on-the-fly repartitioning step). At the same time, having a lot of tiny files is better than having a job constantly fail. This is fixed by adding a high-fanout option (ask every task just to upload their produced files without shuffle/job-level repartitioning). df.writeTo("prod.db.table").option("fanout-enabled", "true").option("distribution-mode", "none").append() Doing multiple uploads in parallel makes everything slow: in my case I've seen a ~0.1% increase on the duration of each task for every new file uploaded in parallel. That means that each task producing 10 files has a 1% slow-down, uploading 100 files in parallel has a 10% slow-down, and uploading 10k files in parallel has a 1000% slow-down. Based on this observation I've seen that it pays off to do the table initialisation with a coarse partitioning schema (such that each tasks writes to a small number of partitions in parallel, around 100 partitions) then add new partitioning columns to the table, and then rewrite the data-files. Repartitioning in Iceberg doesn't have the best algorithms, but improvement-patches exist (see https://github.com/apache/iceberg/issues/16514 and https://github.com/apache/iceberg/issues/16189). Sometimes, the size of the shuffle in the repartitioning-step can be improved by setting the spark-config spark.io.compression.codec=zstd . Later edit: Updated the github links, I've pasted the same link twice.

You can retry the write on conflict if that helps, esp given separate partitions, and it won’t reprocess

You partition in spark? I hear liquid clustering is the new shit rather than partion by and z order

I haven't heard of that before tbh. It looks like a Databricks / Delta Lake feature whereas we use Iceberg. But yes, we partition in Iceberg.

You don't need to use Spark, but I find it does have the best support for Iceberg. If you partition by month you just just be able to run a separate job for each month to load the data without any clashes, just using the overwrite_partitions option. I personally would partition a bit more granular though. If you go by month you will end up with multiple gigabytes per partition. I would just partition by day.

Pyiceberg doesn’t support concurrency yet

Spark does the concurrency well is my understanding.

Where are your parquet files today? In S3? Do you have to use S3 Tables? Because if not you can simply have Iceberg generate its metadata with all of your data in place (no copying required). I know how to do this with Spark - maybe you can do that as a one time thing and then you can go back to using pyiceberg? In fact, I'm pretty sure you can do this with pyiceberg too.

You do not need Spark for this specifically, the commit clashing is the real issue, not the engine. Write all parquet files first without touching the Iceberg metadata, then do a single batch commit using the Iceberg REST catalog or PyIceberg's add_files, which registers existing parquet without rewriting them. Parallelize the file writes, serialize only the final commit. Also agree on the partitioning, id slash month sounds much saner for 30 years of data.

Register the existing parquet files in Iceberg metadata without rewriting the data. Spark has a procedure for this (CALL catalog.system.add_files) and PyIceberg has (Table.add_files). Iceberg partitioning is hidden, it lives in table metadata and not in the S3 directory layout. So create the table partitioned by (id, month(ts)) and register your existing daily files directly, wherever they sit. add_files infers each file's partition value from the parquet footer stats, and each of your daily files already falls inside a single month, which is the only requirement. The slow 10TB rewrite you were fighting with DuckDB/PyIceberg simply goes away. Note that if your current files are small, registration won't fix that, you'll carry the small-file problem into the Iceberg table and will want to compact afterwards (e.g. if you're using Athena - OPTIMIZE ... REWRITE DATA USING BIN_PACK)