DataEng Lab

If you’ve had experience with data lakes, you likely faced significant challenges related to executing updates and deletes. Managing the concurrency between multiple readers and writers, addressing schema evolution in your data, and managing the partitions evolution when data volumes or query patterns change. In this article, we will explore how to use Apache Iceberg with PySpark to address these challenges. To explore Apache Iceberg in more depth, including its architecture and advanced features, see the Complete Course Guide. What is Apache Iceberg? Apache Iceberg is an open table format designed for extensive analytics datasets. It is compatible with widely used big data processing engines such as Apache Spark, Trino, PrestoDB, Flink, and Hive. Iceberg tackles several limitations we listed above by acting as a metadata layer on top of the file format like Apache Parquet and Apache ORC. The following key features of Iceberg effectively address these limitations: Schema Evolution: Allows for seamless schema evolution, overcoming the challenges associated with changes in data structure over time. Transactional Writes: By supporting transactional writes, Iceberg ensures the atomicity, consistency, isolation, and durability (ACID) properties, enhancing data integrity during write operations. Query Isolation: Iceberg provides query isolation, preventing interference between concurrent read and write operations, thus improving overall system reliability and performance. Time Travel: The time travel feature in Iceberg allows users to access historical versions of the data, offering a valuable mechanism for auditing, analysis, and debugging. Partition Pruning: Iceberg’s partition pruning capability optimizes query performance by selectively scanning only relevant partitions, reducing the amount of data processed and improving query speed. Now, let’s start exploring how Iceberg facilitates the implementation of these features when combined with PySpark. Install required dependencies python -m venv iceberg source gx/bin/activate pip install pyspark==3.4.1 Before you start working with Apache Iceberg and PySpark, you need to install the necessary dependencies. Run the commands below to create a virtual environment called iceberg (or choose any name you prefer), activate it, and then install pyspark dependency. Note: If you are using Windows, run the command .icebergScriptsactivate to activate the virtual environment. The versions employed in this article are: Python: 3.11.6 PySpark: 3.4.1 Import required packages Before you can use Apache Iceberg tables in Apache Spark, you must set up the proper integration between them. – Add icerber package to Spark classpath – Configure the catalog 1- Add Iceberg package to Spark Session Before you can use Apache Iceberg tables in Apache Spark, you must set up the proper integration between them. As a first step, you’ll need to specify the required packages to be installed and used with the Spark session. The iceberg-spark-runtime package includes the Iceberg classes that Spark needs to interact with Iceberg tables and metadata. you’re ensuring that these necessary classes are included in the Spark classpath when your Spark shell or application runs 1 2 3 4 5 6 7 8 9 iceberg_spark_jar = 'org.apache.iceberg:iceberg-spark-runtime-3.4_2.12:1.3.0' # Set Iceberg Jar conf = SparkConf() .setAppName("YourAppName") .set('spark.jars.packages', iceberg_spark_jar) # Create spark session spark = SparkSession.builder.config(conf=conf).getOrCreate() 2- Configure the catalog The next important component in the configuration process is the Apache Iceberg catalog. Apache Spark provides an API to add table catalogs, which are utilized for loading, creating, and administering Iceberg tables. This is done by setting the Spark propertyspark.sql.catalog.<catalog-name> with an implementation class for its value. Here  we  defined  a  catalog  named  my_catalog  that  will  be  implemented  using  Iceberg’s implementation of the SparkCatalog class instead of Spark’s default implementation. 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 warehouse_path = "./warehouse" iceberg_spark_jar = 'org.apache.iceberg:iceberg-spark-runtime-3.4_2.12:1.3.0' catalog_name = "demo" # Setup iceberg config conf = SparkConf() .setAppName("YourAppName") .set("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") .set(f"spark.sql.catalog.{catalog_name}", "org.apache.iceberg.spark.SparkCatalog") .set('spark.jars.packages', iceberg_spark_jar) .set(f"spark.sql.catalog.{catalog_name}.warehouse", warehouse_path) .set(f"spark.sql.catalog.{catalog_name}.type", "hadoop") .set("spark.sql.defaultCatalog", catalog_name) # Create spark session spark = SparkSession.builder.config(conf=conf).getOrCreate() To begin working with Iceberg tables in PySpark, it’s essential to configure the PySpark session appropriately. In the following steps, we will use a catalog named demo for tables located under the path ./warehouse of the Hadoop type. Additional configurations can be explored in the Iceberg-Spark-Configuration documentation. Crucially, ensure compatibility between the Iceberg-Spark-Runtime JAR and the PySpark version in use. You can find the necessary JARs in the Iceberg releases. Create and Read an Iceberg Table with PySpark Let’s start by creating and reading an Iceberg table. In the above code, we create a PySpark DataFrame, write it to an Iceberg table, and subsequently display the data stored in the Iceberg table. Now, let’s explore the features that Iceberg comes with to address the issues mentioned in the introduction.  Schema Evolution The flexibility of Data Lakes, allowing storage of diverse data formats, can pose challenges in managing schema changes. Iceberg addresses this by enabling the addition, removal, or modification of table columns without requiring a complete data rewrite. This feature simplifies the process of evolving schemas over time. Let’s modify the previously created table to demonstrate schema evolution. 1 2 3 4 5 6 7 8 spark.sql(f"ALTER TABLE {table_name} RENAME COLUMN job_title TO job") spark.sql(f"ALTER TABLE {table_name} ALTER COLUMN age TYPE bigint") spark.sql(f"ALTER TABLE {table_name} ADD COLUMN salary FLOAT AFTER job") iceberg_df = spark.read.format("iceberg").load(f"{table_name}") iceberg_df.printSchema() iceberg_df.show() spark.sql(f"SELECT * FROM {table_name}.snapshots").show() ACID transactions To demonstrate the ACID with Iceberg table let’s update, add, and delete records from the table. 1 2 3 4 spark.sql(f"UPDATE {table_name} SET salary = 100") spark.sql(f"DELETE FROM {table_name} WHERE age = 42") spark.sql(f"INSERT INTO {table_name} values ('person4', 50, 'Teacher', 2000)") spark.sql(f"SELECT * FROM {table_name}.snapshots").show() In the snapshots table, we can now observe that Iceberg has added three snapshot IDs, each created from the preceding one. If, for any reason, one of the actions fails, the transactions will fail, and the snapshot won’t be created.  ACID transactions Partitioning the table As you may be aware, querying large amounts of data in data lakes can be resource-intensive. Iceberg supports data partitioning by one or more columns. This significantly improves query performance by reducing the volume of data read during queries. 1 2 spark.sql(f"ALTER TABLE {table_name} ADD PARTITION FIELD age") spark.read.format("iceberg").load(f"{table_name}").where("age = 28").show() The code creates a new partition using the age