Comparing various storage formats for object storage compatibility
Originally published on Medium ↗
We have been using parquet storage format for storing our petabyte scale tables in Google Cloud Storage using Hive as our meta store. We encountered some issues with this setup
- Hive writes are always done twice : Hive is not built for object storage, but is built for hadoop like storage engines where renaming files is O(1) operation. But in an object storage like GCS , renaming a folder is an O(n) operation since the contents of that files needs to be copied over to a new folder. There is no O(1) renaming in object storage. Hive relies heavily on renaming folders to offer atomicity. Each new write is written to a staging folder, Once the write is complete, the staging folder will be renamed with the actual folder name. This works great in a traditional file system and a distributed file system , but not in object storage.
- Hive storage is costing twice: Once the hive write to actual folder is complete, hive deletes the staging directory (That is how renaming works in object storage, copy source to target , delete source ). But google started offering soft delete functionality. Which means that whenever the bucket protection(soft delete) is turned on, all the deleted staging folder is going to the soft delete space, the storage in soft delete space is not free either (Think of it as a recycle bin). At one point we were paying more for the data in the soft delete space than the actual storage.
In this article I try to explore data lake formats like Apache Iceberg and Apache Hudi, which claim to be built for object storage.
First let’s reproduce the problem. We will read a hive parquet table, do some transformation and write a new hive parquet table as output. We will keep a tab on how much time each write takes and how many objects are being created etc.
We are doing the following
val df = spark.table("input_table")
df.write.saveAsTable("database.test_output_new")
We will now analyze the output table location in GCS where the contents of the table is written.
Now let’s see the soft delete space below. As you can see the same objects are present in soft delete space as well. This is because spark initially wrote these files into the staging area and then upon commit copies over the files to the final table location. Then it deletes the staging directories, which goes into the soft delete space.
The contents of our sample data is small, But a single new partition write to a big table often contains terabytes of data. Copying over a terabyte of data takes significant amount of time even if it is within the same bucket, We are not only talking about storage cost here but also saving several minutes of time per table write. In some cases the copy from staging directories to actual storage takes 75% of the the total time , as you can see in this SO thread.
Now let’s try the same thing using apache iceberg and see if we are able to solve this problem or not.
Apache Iceberg
Apache iceberg handles atomicity in a different way than hive. There are metatdata files associated with each table. The metadata is updated only after the write is completed. This way even if data is written partially since the data files are not yet part of the metadata the table consistency and atomicity is unaffected. So we are expecting apache iceberg to handle the problems about dual write which we had mentioned at the beginning of this article.
When writing to apache iceberg we wont be writing it as a hive table, but as an iceberg table, Below is the code for that. You should add the spark3 runtime jar to spark shell class path, if you are using spark shell
spark-shell --name spark_session --conf --num-executors 100 --driver-memory 4g --executor-memory 16g --executor-cores 3 —driver-cores 1 --conf spark.sql.broadcastTimeout=-1 --jars gs://da0e807e4a1ef3afe3608218be763e9017a7e38336ce6d6bf110d681a63a0c/application/jars/iceberg-spark-runtime_3.jar
val df = spark.sql("select * from input_table")
import org.apache.iceberg.spark.SparkSchemaUtil
import org.apache.spark.sql._
import org.apache.iceberg.hadoop._
import org.apache.iceberg.types.Types._
import org.apache.iceberg.PartitionSpec
val tables = new HadoopTables(spark.sparkContext.hadoopConfiguration)
val schema = SparkSchemaUtil.convert(df.schema)
val partitionSpec = PartitionSpec.builderFor(schema).build()
val location = "gs://test_iceberg_cbb/data/first_iceberg_table/"
val table = tables.create(schema, partitionSpec, location)
df.write.format("iceberg").mode("overwrite").save("gs://test_iceberg/data/first_iceberg_table/")
Now let’s check the GCS for it’s contents. As you can see below the write is succesful , both data and metadata write is complete.
Now let’s checkout the soft delete space. In the data folder, as you can see there is no object in the soft delete space.
With this small experiment we could conclude that apache iceberg has definitive advantages over apache hive if your data lake is on top of an object storage.