Spark Jobs and Table Storage

Performance of a spark/distributed application depends on how the input tables are stored. Let’s analyze how a spark application’s performance is dependent on the input number of files. We will also check how to store hive tables with the right file size.

Let’s start from the basics. Have you ever read from a data source , processed the data and written to a database/file storage using spark ? Have you observed the output is not always just a single file, but a set of files ? Why is that ? Let’s try doing the same and observe how many files are getting created.

We will read customer profile data from a table stored in gcs and exposed as a hive table, group by customer name, do count and write this information back to gcs. For the first attempt we are disabling adaptive query execution. In the next attempt we will turn it on and see what is the benefit it is bringing.

val df = spark.table("input_table")

val aggregated_df = df.groupBy("first_name").agg(countDistinct("profile_id").alias("first_name_count"))

aggregated_df.printSchema()

spark.conf.set("spark.sql.adaptive.enabled", false)

spark.conf.set("spark.sql.shuffle.partition",100)

aggregated_df.write.format("parquet").save("gs://ajbose-bucket/data/ajbose_test_data_new")

Let’s examine the spark UI.

I have tagged the spark UI with the information that we are going to discuss below. We can observe the following from the spark UI

Understanding the number of tasks in each phase

1. There are three stages in this simple data processing job , the first is to read data from the underlying table’s files. Second is a shuffle to support the wide transformation “groupBy on fullName” . The third is for the distinct operation and writing transformed data to the output table.
2. In the first stage there are 138 tasks. Why are there 138 tasks ? It is an odd number right? Actually the number of tasks in the reading phase for any table is corresponding to the number of files in that table/partition that is being read.
3. In the second and third phase we have exactly 100 tasks , Why is that. It is because we have set the shuffle parts to 100. Whenever spark will have to shuffle the data it will be exactly split into 100 parts( We have turned off Adaptive query execution, which is a feature in spark3. I will write a seperate article about adaptive query execution. )

Any guesses on how many parts will the output table have ? It will be 100 , Why ? Because it was written by 100 different tasks and each task will write a separate file to the output location. But is the file size ideal.

We have too many small files. Too many small files are not good for performance because of the following reasons

1. Overhead of Opening Files: The overhead of opening and closing a large number of files can significantly degrade performance. This is especially true for file systems like HDFS where each file, regardless of its size, requires a block.

2. Increased Garbage Collection: Handling a large number of small Java objects can trigger more frequent garbage collection, which can slow down the processing speed.

3. Network Overhead: Transferring a large number of small files across the network can lead to increased network overhead.

4. Task Scheduling Overhead: Each partition is processed as a separate task. When there are a large number of small files, there are a correspondingly large number of tasks. This leads to increased task scheduling overhead.

5. Data Skew: Small files can often lead to data skew if they are not evenly distributed. Some nodes might be overloaded with data while others have little to do.

6. Inefficient Utilization of Memory and CPU: When Spark reads a small file, it doesn’t fully utilize the memory and CPU, leading to inefficient resource utilization.

To summarize, while writing tables from spark, it is important that wepack the data into right file sizes. How do we do that?

Writing Correctly packaged Tables From Spark

From the earlier example you would have understood that how many parts a table has is directly affected by the shuffle parts configuration if the transformation/processing of data involves a wide transformation, And often all processing involve wide transformation. In that case setting the right number of shuffle partitions before the write to table/file path will result in rightly packaged hive tables with right file sizes.

How do we determine the right number of shuffle parts ? We estimate the size of the data frame and before writing the dataframe to disk we set the shuffleParts according to our esitmates. Below is the code snippets for estimating the shuffle parts.

```scala
import org.apache.spark.sql.DataFrame

def estimateShuffleParts(df: DataFrame): Long = {
// Get the size of the first row
val firstRowSize = df.head().mkString.getBytes.length

// Get the total number of rows
val numRows = df.count()

// Calculate the approximate size
val size = firstRowSize * numRows

// 128 MB files.
size / (1024*1024*128L)
}

val df = spark.read.json("examples/src/main/resources/people.json")

```

Let’s run the same job , but now with the right shuffle parts to see how it is effecting the files in the table.

As you can see this time the files are partitioned better.