Databricks & Iceberg: A Practical Tutorial

Hey guys! Ever heard of Iceberg and wondered how it plays with Databricks? Well, you’re in the right place. This tutorial will walk you through using Iceberg with Databricks, making your data lakehouse dreams a reality. We'll cover everything from setting up your environment to performing common operations. Let's dive in!

What is Apache Iceberg?

Before we get our hands dirty with Databricks, let's quickly cover what Apache Iceberg actually is. Iceberg is an open-source table format for huge analytic datasets. It brings reliability and performance improvements to data lakes. Think of it as a smart layer on top of your data lake that adds features like ACID transactions, schema evolution, time travel, and more. This ensures that your data operations are consistent and efficient, even when dealing with massive datasets.

Why is Iceberg so cool? Well, it solves many of the problems that traditional data lakes face, such as data corruption, inconsistent queries, and the inability to easily evolve schemas. By using Iceberg, you can perform complex operations like upserts and deletes without having to rewrite entire partitions. Plus, its support for time travel allows you to query data as it existed at a specific point in time, which is incredibly useful for auditing and debugging. The integration of Iceberg with Databricks simplifies the creation and management of reliable and high-performance data lakes, which is essential for modern data analytics and machine learning workflows.

Setting up Databricks for Iceberg

Okay, let’s get Databricks ready for some Iceberg action. First, you’ll need a Databricks workspace. If you don’t already have one, head over to the Azure portal or AWS Marketplace and create a new Databricks workspace. Once you have your workspace up and running, you’ll need to configure it to work with Iceberg. This involves setting up the necessary libraries and configurations.

To start, create a new cluster in Databricks. When creating the cluster, make sure you select a Databricks Runtime version that supports Iceberg. As of now, Databricks Runtime 10.4 LTS and later versions come with built-in support for Iceberg. If you're using an older runtime, you'll need to install the Iceberg library manually. To do this, you can add the Iceberg JAR file as a library to your cluster. You can download the latest Iceberg JAR from the Apache Iceberg website or Maven Central. Once you've added the library, you'll need to configure Spark to use the Iceberg catalog. This involves setting the spark.sql.catalog.your_catalog_name property to org.apache.iceberg.spark.SparkCatalog and configuring the catalog properties, such as the warehouse location. For example, you might set spark.sql.catalog.iceberg_catalog.warehouse to a location in your cloud storage where Iceberg will store the table metadata. Finally, you'll want to enable Iceberg's support for time travel by setting spark.sql.extensions to org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions. This will allow you to query your data as it existed at a specific point in time. With these configurations in place, your Databricks environment will be fully equipped to leverage the power of Iceberg for your data lakehouse projects.

Configure Spark

Next, you need to configure Spark to work with Iceberg. Add the following configurations to your Spark session:

spark.conf.set("spark.sql.catalog.iceberg_catalog", "org.apache.iceberg.spark.SparkCatalog")
spark.conf.set("spark.sql.catalog.iceberg_catalog.warehouse", "dbfs:/iceberg_warehouse")
spark.conf.set("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")

Replace dbfs:/iceberg_warehouse with the actual path where you want to store your Iceberg tables.

Configuring Spark is essential for integrating Iceberg into your Databricks environment. The first step involves setting the spark.sql.catalog.iceberg_catalog property to org.apache.iceberg.spark.SparkCatalog. This tells Spark to use Iceberg as the catalog for managing your tables. Next, you need to specify the location where Iceberg will store the table metadata. This is done by setting the spark.sql.catalog.iceberg_catalog.warehouse property to a path in your cloud storage, such as dbfs:/iceberg_warehouse. Make sure this location is accessible to your Databricks cluster. Additionally, it's crucial to enable Iceberg's support for time travel by setting spark.sql.extensions to org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions. This allows you to query your data as it existed at a specific point in time, providing powerful auditing and debugging capabilities. By properly configuring Spark with these settings, you ensure that Iceberg can seamlessly manage your data lakehouse, providing features like ACID transactions, schema evolution, and efficient query performance. These configurations are fundamental to unlocking the full potential of Iceberg in your Databricks environment, enabling you to build reliable and scalable data pipelines.

Creating Your First Iceberg Table

Alright, now for the fun part: creating your first Iceberg table! You can create an Iceberg table using SQL or the DataFrame API. Here’s how to do it with SQL:

CREATE TABLE iceberg_catalog.my_database.my_table (
 id BIGINT,
 name STRING,
 ts TIMESTAMP
) USING iceberg

And here’s the DataFrame API equivalent:

data = [ (1, "Alice", "2024-01-01 00:00:00"), (2, "Bob", "2024-01-02 00:00:00") ]
df = spark.createDataFrame(data, ["id", "name", "ts"])
df.writeTo("iceberg_catalog.my_database.my_table").create()

Creating your first Iceberg table is a straightforward process that can be accomplished using either SQL or the DataFrame API. When using SQL, you define the schema of your table, including the column names and data types, and specify USING iceberg to indicate that you want to create an Iceberg table. For example, you might create a table named my_table in the my_database database with columns for id, name, and ts. Alternatively, you can use the DataFrame API to create an Iceberg table from a DataFrame. This involves creating a DataFrame with the desired data and schema, and then using the writeTo method to specify the table name and the create method to create the table. Both methods achieve the same result: a new Iceberg table that you can start using to store and manage your data. Whether you prefer SQL or the DataFrame API, Iceberg provides a flexible and easy-to-use interface for creating tables in your data lakehouse.

Writing Data to the Iceberg Table

Time to put some data into our new Iceberg table. You can insert data using SQL like this:

INSERT INTO iceberg_catalog.my_database.my_table VALUES (3, 'Charlie', '2024-01-03 00:00:00')

Or using the DataFrame API:

data = [(4, "David", "2024-01-04 00:00:00")]
df = spark.createDataFrame(data, ["id", "name", "ts"])
df.writeTo("iceberg_catalog.my_database.my_table").append()

Writing data to your Iceberg table is a fundamental operation that can be performed using either SQL or the DataFrame API. When using SQL, you can insert data into the table using the INSERT INTO statement, specifying the values for each column. For example, you might insert a new row with an id of 3, a name of 'Charlie', and a timestamp of '2024-01-03 00:00:00'. Alternatively, you can use the DataFrame API to write data to the table by creating a DataFrame with the data you want to insert and then using the writeTo method to specify the table name and the append method to add the data to the table. The append method is particularly useful for adding new data to an existing table without overwriting the existing data. Both methods ensure that the data is written to the Iceberg table in a consistent and reliable manner, taking advantage of Iceberg's ACID transaction capabilities. Whether you prefer SQL or the DataFrame API, Iceberg provides a flexible and efficient way to write data to your tables, enabling you to build robust and scalable data pipelines.

Querying the Iceberg Table

Now, let’s see how to query the data we just inserted. A simple SELECT statement will do the trick:

SELECT * FROM iceberg_catalog.my_database.my_table

Querying your Iceberg table is as simple as using a standard SQL SELECT statement. You can retrieve all columns and rows from the table using SELECT * FROM iceberg_catalog.my_database.my_table. This will return the data you've inserted, allowing you to verify that the data has been written correctly. You can also use more complex SQL queries to filter, sort, and aggregate the data as needed. Iceberg's integration with Spark SQL ensures that these queries are executed efficiently, taking advantage of Spark's distributed processing capabilities. Whether you're performing simple data retrieval or complex analytical queries, Iceberg provides a seamless and performant querying experience, making it easy to access and analyze your data lakehouse.

Time Travel with Iceberg

One of the coolest features of Iceberg is time travel. You can query the table as it existed at a specific point in time. First, find the snapshot ID:

SELECT * FROM iceberg_catalog.my_database.my_table.history

Then, use the snapshot ID to query the table:

SELECT * FROM iceberg_catalog.my_database.my_table VERSION AS OF snapshot_id

Replace snapshot_id with the actual snapshot ID you retrieved from the history table.

Time travel is a powerful feature of Iceberg that allows you to query the table as it existed at a specific point in time. This is incredibly useful for auditing, debugging, and reproducing results. To use time travel, you first need to find the snapshot ID of the table at the point in time you're interested in. You can do this by querying the history table associated with your Iceberg table. The history table contains information about all the snapshots of the table, including the snapshot ID, the timestamp of the snapshot, and the operation that created the snapshot. Once you have the snapshot ID, you can use the VERSION AS OF clause in your SQL query to specify that you want to query the table as it existed at that snapshot. For example, if you want to query the table as it existed at snapshot ID 12345, you would use the query SELECT * FROM iceberg_catalog.my_database.my_table VERSION AS OF 12345. This will return the data that was in the table at that point in time, allowing you to analyze historical data and track changes over time. Time travel provides a unique and valuable capability for managing and understanding your data lakehouse.

Schema Evolution

Iceberg makes schema evolution a breeze. Let’s add a new column to our table:

ALTER TABLE iceberg_catalog.my_database.my_table ADD COLUMN age INT

Now you can write data to the new column without any hassle.

Schema evolution is a critical feature for any data lakehouse, and Iceberg makes it incredibly easy to manage changes to your table schemas. With Iceberg, you can add, drop, or rename columns without having to rewrite the entire table. This is a significant advantage over traditional data lake formats, which often require complex and time-consuming migrations to handle schema changes. To add a new column to your Iceberg table, you can use the ALTER TABLE statement with the ADD COLUMN clause. For example, to add an age column of type INT to your table, you would use the statement ALTER TABLE iceberg_catalog.my_database.my_table ADD COLUMN age INT. Once you've added the column, you can start writing data to it without any disruption to your existing data. Iceberg automatically handles the schema change, ensuring that your data remains consistent and accessible. This makes it easy to adapt your data lakehouse to evolving business requirements and new data sources, without the risk of data corruption or downtime. Schema evolution in Iceberg is a game-changer for data lake management, providing the flexibility and agility you need to stay ahead in today's fast-paced data landscape.

Cleaning Up

Finally, let’s clean up our resources. Drop the table when you’re done:

DROP TABLE iceberg_catalog.my_database.my_table

And that’s it! You’ve successfully used Iceberg with Databricks.

Cleaning up your resources after you're done is an important step in managing your data lakehouse. To remove the Iceberg table you created, you can use the DROP TABLE statement. For example, to drop the table my_table in the my_database database, you would use the statement DROP TABLE iceberg_catalog.my_database.my_table. This will remove the table and all its associated metadata from the Iceberg catalog. It's a good practice to drop tables that you no longer need to avoid cluttering your environment and consuming unnecessary storage resources. Dropping a table in Iceberg is a simple and efficient operation, ensuring that you can easily manage your data lakehouse and keep it organized.

Conclusion

So there you have it! Using Iceberg with Databricks is a powerful way to manage your data lake. You can perform all sorts of advanced operations, like time travel and schema evolution, with ease. Go ahead and explore more features and integrations. Happy data crunching!