Prerequisites:

1. Azure Blob Storage Account: Ensure that you have access to the Azure Blob Storage account and the necessary permissions to read data from the containers.
2. Apache Iceberg Table: Set up an Apache Iceberg catalog or know the location where you want to create Iceberg tables.
3. Apache Spark Environment: Have a Spark environment set up with the necessary dependencies for Iceberg.
4. Iceberg Spark Runtime Jar: Download the appropriate Apache Iceberg Spark runtime JAR file that matches your Spark version.
5. Azure Storage SDK: Include the Azure Storage SDK in your Spark environment for accessing Blob Storage.
6. Hadoop-Azure Module: Include the Hadoop-Azure module in your Spark environment to allow Hadoop to work with Azure Blob Storage.

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
 .appName("AzureBlobToIceberg")
 .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark3-runtime:VERSION")
 .config("spark.hadoop.fs.azure", "org.apache.hadoop.fs.azure.NativeAzureFileSystem")
 .config("spark.hadoop.fs.azure.account.key.YOUR_STORAGE_ACCOUNT.blob.core.windows.net", "YOUR_STORAGE_ACCOUNT_ACCESS_KEY")
 .getOrCreate()


Replace VERSION with the correct version of the Iceberg Spark runtime library, YOUR_STORAGE_ACCOUNT with your Azure Blob Storage account name, and YOUR_STORAGE_ACCOUNT_ACCESS_KEY with your Azure Storage account access key.

val df = spark.read.format("FORMAT")
 .load("wasbs://CONTAINER@YOUR_STORAGE_ACCOUNT.blob.core.windows.net/PATH_TO_YOUR_DATA")


Replace FORMAT with the format of your data (e.g., csv, parquet, json), CONTAINER with the name of your Blob Storage container, and PATH_TO_YOUR_DATA with the path to your data within the container.

If you haven’t already configured an Iceberg catalog in your Spark environment, you’ll need to do so:

spark.conf.set("spark.sql.catalog.my_catalog", "org.apache.iceberg.spark.SparkCatalog")
spark.conf.set("spark.sql.catalog.my_catalog.type", "hadoop")
spark.conf.set("spark.sql.catalog.my_catalog.warehouse", "hdfs://PATH_TO_ICEBERG_WAREHOUSE")


Replace PATH_TO_ICEBERG_WAREHOUSE with the path to your Iceberg warehouse directory.

df.write.format("iceberg")
 .option("write-format", "parquet") // Or your desired format
 .mode("append") // Use "overwrite" if you want to replace existing data
 .save("my_catalog.db.TABLE_NAME")


Replace TABLE_NAME with the name of the Iceberg table you want to write to. If the table doesn’t exist, Iceberg will create it.

After the write operation, you can verify that the data has been transferred successfully:

val icebergDf = spark.read.format("iceberg")
 .load("my_catalog.db.TABLE_NAME")

icebergDf.show()


This will display a few rows from the Iceberg table.

Notes:

• Ensure that the Spark version you are using is compatible with the Iceberg version.
• You may need to add additional configurations depending on your specific setup and security requirements.
• If you are dealing with large datasets, consider using Spark’s partitioning and bucketing features to optimize the data transfer and subsequent queries.
• Always test your data pipeline with a subset of data before running it on the full dataset to ensure that everything is configured correctly.