Comparing Performance of Big Data File Formats: A Practical Guide | by Sarthak Sarbahi | Jan, 2024

Surroundings setup

On this information, we’re going to make use of JupyterLab with Docker and MinIO. Consider Docker as a useful software that simplifies working functions, and MinIO as a versatile storage resolution excellent for dealing with plenty of several types of knowledge. Right here’s how we’ll set issues up:

I’m not diving deep into each step right here since there’s already a terrific tutorial for that. I counsel checking it out first, then coming again to proceed with this one.

As soon as the whole lot’s prepared, we’ll begin by getting ready our pattern knowledge. Open a brand new Jupyter pocket book to start.

First up, we have to set up the s3fs Python package deal, important for working with MinIO in Python.

!pip set up s3fs

Following that, we’ll import the required dependencies and modules.

import os
import s3fs
import pyspark
from pyspark.sql import SparkSession
from pyspark import SparkContext
import pyspark.sql.capabilities as F
from pyspark.sql import Row
import pyspark.sql.sorts as T
import datetime
import time

We’ll additionally set some setting variables that shall be helpful when interacting with MinIO.

# Outline setting variables
os.environ["MINIO_KEY"] = "minio"
os.environ["MINIO_SECRET"] = "minio123"
os.environ["MINIO_ENDPOINT"] = "http://minio1:9000"

Then, we’ll arrange our Spark session with the required settings.

# Create Spark session
spark = SparkSession.builder 
.appName("big_data_file_formats") 
.config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.11.1026,org.apache.spark:spark-avro_2.12:3.5.0,io.delta:delta-spark_2.12:3.0.0") 
.config("spark.hadoop.fs.s3a.endpoint", os.environ["MINIO_ENDPOINT"]) 
.config("spark.hadoop.fs.s3a.entry.key", os.environ["MINIO_KEY"]) 
.config("spark.hadoop.fs.s3a.secret.key", os.environ["MINIO_SECRET"]) 
.config("spark.hadoop.fs.s3a.path.type.entry", "true") 
.config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") 
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") 
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") 
.enableHiveSupport() 
.getOrCreate()

Let’s simplify this to grasp it higher.

• spark.jars.packages: Downloads the required JAR information from the Maven repository. A Maven repository is a central place used for storing construct artifacts like JAR information, libraries, and different dependencies which might be utilized in Maven-based initiatives.
• spark.hadoop.fs.s3a.endpoint: That is the endpoint URL for MinIO.
• spark.hadoop.fs.s3a.entry.key and spark.hadoop.fs.s3a.secret.key: That is the entry key and secret key for MinIO. Word that it’s the similar because the username and password used to entry the MinIO net interface.
• spark.hadoop.fs.s3a.path.type.entry: It’s set to true to allow path-style entry for the MinIO bucket.
• spark.hadoop.fs.s3a.impl: That is the implementation class for S3A file system.
• spark.sql.extensions: Registers Delta Lake’s SQL instructions and configurations inside the Spark SQL parser.
• spark.sql.catalog.spark_catalog: Units the Spark catalog to Delta Lake’s catalog, permitting desk administration and metadata operations to be dealt with by Delta Lake.

Choosing the proper JAR model is essential to keep away from errors. Utilizing the identical Docker picture, the JAR model talked about right here ought to work high quality. When you encounter setup points, be happy to go away a remark. I’ll do my greatest to help you 🙂

Our subsequent step is to create an enormous Spark dataframe. It’ll have 10 million rows, divided into ten columns — half are textual content, and half are numbers.

# Generate pattern knowledge
num_rows = 10000000
df = spark.vary(0, num_rows)
# Add columns
for i in vary(1, 10):  # Since we have already got one column
if i % 2 == 0:
# Integer column
df = df.withColumn(f"int_col_{i}", (F.randn() * 100).solid(T.IntegerType()))
else:
# String column
df = df.withColumn(f"str_col_{i}", (F.rand() * num_rows).solid(T.IntegerType()).solid("string"))
df.rely()

Let’s peek on the first few entries to see what they appear like.

# Present rows from pattern knowledge
df.present(10,truncate = False)
+---+---------+---------+---------+---------+---------+---------+---------+---------+---------+
|id |str_col_1|int_col_2|str_col_3|int_col_4|str_col_5|int_col_6|str_col_7|int_col_8|str_col_9|
+---+---------+---------+---------+---------+---------+---------+---------+---------+---------+
|0  |7764018  |128      |1632029  |-15      |5858297  |114      |1025493  |-88      |7376083  |
|1  |2618524  |118      |912383   |235      |6684042  |-115     |9882176  |170      |3220749  |
|2  |6351000  |75       |3515510  |26       |2605886  |89       |3217428  |87       |4045983  |
|3  |4346827  |-70      |2627979  |-23      |9543505  |69       |2421674  |-141     |7049734  |
|4  |9458796  |-106     |6374672  |-142     |5550170  |25       |4842269  |-97      |5265771  |
|5  |9203992  |23       |4818602  |42       |530044   |28       |5560538  |-75      |2307858  |
|6  |8900698  |-130     |2735238  |-135     |1308929  |22       |3279458  |-22      |3412851  |
|7  |6876605  |-35      |6690534  |-41      |273737   |-178     |8789689  |88       |4200849  |
|8  |3274838  |-42      |1270841  |-62      |4592242  |133      |4665549  |-125     |3993964  |
|9  |4904488  |206      |2176042  |58       |1388630  |-63      |9364695  |78       |2657371  |
+---+---------+---------+---------+---------+---------+---------+---------+---------+---------+
solely exhibiting high 10 rows

To know the construction of our dataframe, we’ll use df.printSchema() to see the forms of knowledge it incorporates. After this, we’ll create 4 CSV information. These shall be used for Parquet, Avro, ORC, and Delta Lake. We’re doing this to keep away from any bias in efficiency testing — utilizing the identical CSV lets Spark cache and optimize issues within the background.

# Write 4 CSVs for evaluating efficiency for each file sort
df.write.csv("s3a://mybucket/ten_million_parquet.csv")
df.write.csv("s3a://mybucket/ten_million_avro.csv")
df.write.csv("s3a://mybucket/ten_million_orc.csv")
df.write.csv("s3a://mybucket/ten_million_delta.csv")

Now, we’ll make 4 separate dataframes from these CSVs, every one for a unique file format.

# Learn all 4 CSVs to create dataframes
schema = T.StructType([
T.StructField("id", T.LongType(), nullable=False),
T.StructField("str_col_1", T.StringType(), nullable=True),
T.StructField("int_col_2", T.IntegerType(), nullable=True),
T.StructField("str_col_3", T.StringType(), nullable=True),
T.StructField("int_col_4", T.IntegerType(), nullable=True),
T.StructField("str_col_5", T.StringType(), nullable=True),
T.StructField("int_col_6", T.IntegerType(), nullable=True),
T.StructField("str_col_7", T.StringType(), nullable=True),
T.StructField("int_col_8", T.IntegerType(), nullable=True),
T.StructField("str_col_9", T.StringType(), nullable=True)
])
df_csv_parquet = spark.learn.format("csv").possibility("header",True).schema(schema).load("s3a://mybucket/ten_million_parquet.csv")
df_csv_avro = spark.learn.format("csv").possibility("header",True).schema(schema).load("s3a://mybucket/ten_million_avro.csv")
df_csv_orc = spark.learn.format("csv").possibility("header",True).schema(schema).load("s3a://mybucket/ten_million_orc.csv")
df_csv_delta = spark.learn.format("csv").possibility("header",True).schema(schema).load("s3a://mybucket/ten_million_delta.csv")

And that’s it! We’re all set to discover these huge knowledge file codecs.

Working with Parquet

Parquet is a column-oriented file format that meshes very well with Apache Spark, making it a best choice for dealing with huge knowledge. It shines in analytical situations, significantly while you’re sifting by way of knowledge column by column.

One among its neat options is the power to retailer knowledge in a compressed format, with snappy compression being the go-to alternative. This not solely saves house but in addition enhances efficiency.

One other cool facet of Parquet is its versatile strategy to knowledge schemas. You can begin off with a primary construction after which easily develop by including extra columns as your wants develop. This adaptability makes it tremendous user-friendly for evolving knowledge initiatives.

Now that we’ve received a deal with on Parquet, let’s put it to the check. We’re going to jot down 10 million information right into a Parquet file and control how lengthy it takes. As an alternative of utilizing the %timeit Python perform, which runs a number of instances and could be heavy on sources for giant knowledge duties, we’ll simply measure it as soon as.

# Write knowledge as Parquet
start_time = time.time()
df_csv_parquet.write.parquet("s3a://mybucket/ten_million_parquet2.parquet")
end_time = time.time()
print(f"Time taken to jot down as Parquet: {end_time - start_time} seconds")

For me, this activity took 15.14 seconds, however keep in mind, this time can change relying in your laptop. For instance, on a much less highly effective PC, it took longer. So, don’t sweat it in case your time is totally different. What’s necessary right here is evaluating the efficiency throughout totally different file codecs.

Subsequent up, we’ll run an aggregation question on our Parquet knowledge.

# Perfom aggregation question utilizing Parquet knowledge
start_time = time.time()
df_parquet = spark.learn.parquet("s3a://mybucket/ten_million_parquet2.parquet")
df_parquet 
.choose("str_col_5","str_col_7","int_col_2") 
.groupBy("str_col_5","str_col_7") 
.rely() 
.orderBy("rely") 
.restrict(1) 
.present(truncate = False)
end_time = time.time()
print(f"Time taken for question: {end_time - start_time} seconds")
+---------+---------+-----+
|str_col_5|str_col_7|rely|
+---------+---------+-----+
|1        |6429997  |1    |
+---------+---------+-----+

This question completed in 12.33 seconds. Alright, now let’s swap gears and discover the ORC file format.

Working with ORC

The ORC file format, one other column-oriented contender, may not be as well-known as Parquet, but it surely has its personal perks. One standout characteristic is its capability to compress knowledge much more successfully than Parquet, whereas utilizing the identical snappy compression algorithm.

It’s a success within the Hive world, because of its help for ACID operations in Hive tables. ORC can also be tailored for dealing with giant streaming reads effectively.

Plus, it’s simply as versatile as Parquet on the subject of schemas — you may start with a primary construction after which add extra columns as your venture grows. This makes ORC a sturdy alternative for evolving huge knowledge wants.

Let’s dive into testing ORC’s writing efficiency.

# Write knowledge as ORC
start_time = time.time()
df_csv_orc.write.orc("s3a://mybucket/ten_million_orc2.orc")
end_time = time.time()
print(f"Time taken to jot down as ORC: {end_time - start_time} seconds")

It took me 12.94 seconds to finish the duty. One other focal point is the scale of the info written to the MinIO bucket. Within the ten_million_orc2.orc folder, you’ll discover a number of partition information, every of a constant dimension. Each partition ORC file is about 22.3 MiB, and there are 16 information in whole.

Evaluating this to Parquet, every Parquet partition file is round 26.8 MiB, additionally totaling 16 information. This reveals that ORC certainly gives higher compression than Parquet.

Subsequent, we’ll check how ORC handles an aggregation question. We’re utilizing the identical question for all file codecs to maintain our benchmarking truthful.

# Carry out aggregation utilizing ORC knowledge
df_orc = spark.learn.orc("s3a://mybucket/ten_million_orc2.orc")
start_time = time.time()
df_orc 
.choose("str_col_5","str_col_7","int_col_2") 
.groupBy("str_col_5","str_col_7") 
.rely() 
.orderBy("rely") 
.restrict(1) 
.present(truncate = False)
end_time = time.time()
print(f"Time taken for question: {end_time - start_time} seconds")
+---------+---------+-----+
|str_col_5|str_col_7|rely|
+---------+---------+-----+
|1        |2906292  |1    |
+---------+---------+-----+

The ORC question completed in 13.44 seconds, a tad longer than Parquet’s time. With ORC checked off our record, let’s transfer on to experimenting with Avro.

Working with Avro

Avro is a row-based file format with its personal distinctive strengths. Whereas it doesn’t compress knowledge as effectively as Parquet or ORC, it makes up for this with a quicker writing pace.

What actually units Avro aside is its wonderful schema evolution capabilities. It handles modifications like added, eliminated, or altered fields with ease, making it a go-to alternative for situations the place knowledge buildings evolve over time.

Avro is especially well-suited for workloads that contain numerous knowledge writing.

Now, let’s try how Avro does with writing knowledge.

# Write knowledge as Avro
start_time = time.time()
df_csv_avro.write.format("avro").save("s3a://mybucket/ten_million_avro2.avro")
end_time = time.time()
print(f"Time taken to jot down as Avro: {end_time - start_time} seconds")

It took me 12.81 seconds, which is definitely faster than each Parquet and ORC. Subsequent, we’ll have a look at Avro’s efficiency with an aggregation question.

# Carry out aggregation utilizing Avro knowledge
df_avro = spark.learn.format("avro").load("s3a://mybucket/ten_million_avro2.avro")
start_time = time.time()
df_avro 
.choose("str_col_5","str_col_7","int_col_2") 
.groupBy("str_col_5","str_col_7") 
.rely() 
.orderBy("rely") 
.restrict(1) 
.present(truncate = False)
end_time = time.time()
print(f"Time taken for question: {end_time - start_time} seconds")
+---------+---------+-----+
|str_col_5|str_col_7|rely|
+---------+---------+-----+
|1        |6429997  |1    |
+---------+---------+-----+

This question took about 15.42 seconds. So, on the subject of querying, Parquet and ORC are forward when it comes to pace. Alright, it’s time to discover our remaining and latest file format — Delta Lake.

Working with Delta Lake

Delta Lake is a brand new star within the huge knowledge file format universe, intently associated to Parquet when it comes to storage dimension — it’s like Parquet however with some further options.

When writing knowledge, Delta Lake takes a bit longer than Parquet, principally due to its _delta_log folder, which is vital to its superior capabilities. These capabilities embody ACID compliance for dependable transactions, time journey for accessing historic knowledge, and small file compaction to maintain issues tidy.

Whereas it’s a newcomer within the huge knowledge scene, Delta Lake has shortly turn into a favourite on cloud platforms that run Spark, outpacing its use in on-premises techniques.

Let’s transfer on to testing Delta Lake’s efficiency, beginning with an information writing check.

# Write knowledge as Delta
start_time = time.time()
df_csv_delta.write.format("delta").save("s3a://mybucket/ten_million_delta2.delta")
end_time = time.time()
print(f"Time taken to jot down as Delta Lake: {end_time - start_time} seconds")

The write operation took 17.78 seconds, which is a bit longer than the opposite file codecs we’ve checked out. A neat factor to note is that within the ten_million_delta2.delta folder, every partition file is definitely a Parquet file, comparable in dimension to what we noticed with Parquet. Plus, there’s the _delta_log folder.

The _delta_log folder within the Delta Lake file format performs a important function in how Delta Lake manages and maintains knowledge integrity and versioning. It is a key element that units Delta Lake aside from different huge knowledge file codecs. This is a easy breakdown of its perform:

1. Transaction Log: The _delta_log folder incorporates a transaction log that information each change made to the info within the Delta desk. This log is a collection of JSON information that element the additions, deletions, and modifications to the info. It acts like a complete diary of all the info transactions.
2. ACID Compliance: This log permits ACID (Atomicity, Consistency, Isolation, Sturdiness) compliance. Each transaction in Delta Lake, like writing new knowledge or modifying present knowledge, is atomic and constant, guaranteeing knowledge integrity and reliability.
3. Time Journey and Auditing: The transaction log permits for “time journey”, which implies you may simply view and restore earlier variations of the info. That is extraordinarily helpful for knowledge restoration, auditing, and understanding how knowledge has advanced over time.
4. Schema Enforcement and Evolution: The _delta_log additionally retains monitor of the schema (construction) of the info. It enforces the schema throughout knowledge writes and permits for protected evolution of the schema over time with out corrupting the info.
5. Concurrency and Merge Operations: It manages concurrent reads and writes, guaranteeing that a number of customers can entry and modify the info on the similar time with out conflicts. This makes it preferrred for complicated operations like merge, replace, and delete.

In abstract, the _delta_log folder is the mind behind Delta Lake’s superior knowledge administration options, providing sturdy transaction logging, model management, and reliability enhancements that aren’t usually obtainable in less complicated file codecs like Parquet or ORC.

Now, it’s time to see how Delta Lake fares with an aggregation question.

# Carry out aggregation utilizing Delta knowledge
df_delta = spark.learn.format("delta").load("s3a://mybucket/ten_million_delta2.delta")
start_time = time.time()
df_delta 
.choose("str_col_5","str_col_7","int_col_2") 
.groupBy("str_col_5","str_col_7") 
.rely() 
.orderBy("rely") 
.restrict(1) 
.present(truncate = False)
end_time = time.time()
print(f"Time taken for question: {end_time - start_time} seconds")
+---------+---------+-----+
|str_col_5|str_col_7|rely|
+---------+---------+-----+
|1        |2906292  |1    |
+---------+---------+-----+

This question completed in about 15.51 seconds. Whereas it is a tad slower in comparison with Parquet and ORC, it’s fairly shut. It means that Delta Lake’s efficiency in real-world situations is sort of much like that of Parquet.

Superior! We’ve wrapped up all our experiments. Let’s recap our findings within the subsequent part.

When to make use of which file format?

We’ve wrapped up our testing, so let’s convey all our findings collectively. For knowledge writing, Avro takes the highest spot. That’s actually what it’s greatest at in sensible situations.

On the subject of studying and working aggregation queries, Parquet leads the pack. Nonetheless, this doesn’t imply ORC and Delta Lake fall brief. As columnar file codecs, they carry out admirably in most conditions.

Right here’s a fast rundown:

• Select ORC for the very best compression, particularly when you’re utilizing Hive and Pig for analytical duties.
• Working with Spark? Parquet and Delta Lake are your go-to selections.
• For situations with plenty of knowledge writing, like touchdown zone areas, Avro is the very best match.

And that’s a wrap on this tutorial!