Can We Predict How Many Spark Tasks an Iceberg Ingestion Will Create?

I have recently started looking at how Apache Spark(Dataproc) works with Iceberg as a data lake, with a view to better understanding it from an overall systems perspective.

In a series of blog posts, I will cover aspects such as workload modelling, task estimation, queues, and provide an initial overview of how I believe this works for data ingestion and overall systems performance.

There have been a few posts on similar topics; however, this is not another recommendation to divide the total input size by 128 MiB. The estimator uses the individual file sizes, reproduces Spark’s file-packing behaviour, and then compares the result with the task count observed in production.

Tasks

A Spark task is a single unit of work within a stage. By estimating how many tasks a stage will create, we can compare that workload with the number of available executor task slots. This allows us to estimate how many tasks can run concurrently and how many execution waves will be required to complete the stage.

This raises an important question: can we effectively estimate the number of tasks Spark will create for a given input dataset before running the job?

Partitions

Initially, we need to understand the file sizes to be ingested into our data lake. We can find these on GCP using the following command; this has been split for easy reading.

# gcloud storage ls -L gs://<your-bucket>/ > input_data 
# awk '/Content-Length:/{print $2}' input_data > data

Once we have the data in our file, we can look at it in different ways. One way is to look at the distribution of the data, which in our example is the count and size in MB for our files.

# awk '{printf("%d \n", $1/1048576)}' data |sort|uniq -c |sort
1 13
1 41
2 12
3 17
3 22
5 32
6 16
12 0
19 75
26 60
30 57
34 61
35 59
49 40
51 38
94 64
95 33
97 58
100 30
105 25
119 62
136 23
150 35
155 56
187 55
197 24
213 51
215 46
225 48
225 52
235 50
237 63
240 65
291 54
293 53
334 42
496 49
617 43
620 66
746 47
848 44
1220 45
2179 39

Estimating Spark Tasks: 128 MiB Partitions and 4 MiB File-Open Cost

For file-based reads, Spark attempts to combine input files into partitions with a default maximum size of 128 MiB. When calculating whether files can be placed in the same partition, Spark also adds a 4 MiB file-open cost for every file or file split.

This overhead does not represent data that Spark physically reads; it is a planning value intended to account for the cost of opening many small files.

Spark subsequently splits files larger than 128 MiB, sorts the resulting chunks from largest to smallest, and then packs them into partitions.

Each completed partition normally becomes one task in the input scan stage.

partition_size = 128 MiB
open_cost = 4 MiB

chunks = split every input file into pieces no larger than partition_size
sort chunks from largest to smallest

task_count = 0
current_partition_size = 0

for each chunk:

    if current_partition_size + chunk.size > partition_size:
        task_count = task_count + 1
        current_partition_size = 0

    current_partition_size =
        current_partition_size
        + chunk.size
        + open_cost

if current_partition_size > 0:
    task_count = task_count + 1

return task_count

For example, two 50 MiB files have a planning size of approximately 108 MiB after adding the 4 MiB cost for each file, so they fit into one 128 MiB partition. Three such files would require approximately 162 MiB, so Spark would place them into more than one partition.

Spark input task estimate
================================================
Input files                 : 10,989
Total input bytes           : 534,473,867,264
Total input size            : 509,714.00 MiB
Maximum partition bytes     : 134,217,728
Maximum partition size      : 128.00 MiB
File-open overhead          : 4,194,304 bytes per file/split
Total planning size         : 553,670.00 MiB
------------------------------------------------
Estimated Spark tasks       : 5,540
Average data per task       : 92.01 MiB
Smallest task               : 12.00 MiB
Largest task                : 124.00 MiB
Average chunks per task     : 1.98
Minimum chunks in a task    : 1
Maximum chunks in a task    : 14

Above, we can see the output of our estimation tool. We had 10989 files sizes input and estimated 5,540 tasks; when the job ran, 5403 tasks were observed, so a difference of 137.

This is pretty close, roughly within 2.5%, and is a good way to understand the sizing of clusters for processing data ingestion into the iceberg. It also allows for future runtime estimation, which we will cover in a future post.