Feat: Support Iceberg format in the local Airflow setup (SQLMesh#1755)

Author: izeigerman

examples/airflow/Dockerfile.template

@@ -1,4 +1,4 @@
-FROM apache/spark-py:latest AS spark
+FROM apache/spark-py:v3.4.0 AS spark
 
 FROM apache/airflow:$AIRFLOW_VERSION
 
@@ -22,8 +22,9 @@ RUN chown -R airflow /opt/spark
 ENV SPARK_HOME="/opt/spark"
 ENV PATH="$PATH:$SPARK_HOME/bin"
 
-# Install Postgres driver for Spark
-RUN curl https://jdbc.postgresql.org/download/postgresql-42.5.0.jar -o /opt/spark/jars/postgresql-42.5.0.jar
+# Install Postgres driver and Iceberg for Spark
+RUN curl https://jdbc.postgresql.org/download/postgresql-42.5.0.jar -o /opt/spark/jars/postgresql-42.5.0.jar && \
+    curl -L https://search.maven.org/remotecontent?filepath=org/apache/iceberg/iceberg-spark-runtime-3.4_2.12/1.4.2/iceberg-spark-runtime-3.4_2.12-1.4.2.jar -o /opt/spark/jars/iceberg-spark-runtime-3.4_2.12-1.4.2.jar
 
 # Install Hadoop
 RUN curl https://dlcdn.apache.org/hadoop/common/hadoop-3.3.4/hadoop-3.3.4.tar.gz -o hadoop-3.3.4.tar.gz && \

examples/airflow/spark_conf/spark-defaults.conf

@@ -1,3 +1,7 @@
-spark.hadoop.hive.exec.dynamic.partition true
-spark.hadoop.hive.exec.dynamic.partition.mode nonstrict
-spark.sql.sources.partitionOverwriteMode dynamic
+spark.hadoop.hive.exec.dynamic.partition             true
+spark.hadoop.hive.exec.dynamic.partition.mode        nonstrict
+spark.sql.sources.partitionOverwriteMode             dynamic
+
+spark.sql.extensions                                 org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
+spark.sql.catalog.spark_catalog                      org.apache.iceberg.spark.SparkSessionCatalog
+spark.sql.catalog.spark_catalog.type                 hive

examples/sushi/config.py

@@ -54,15 +54,15 @@
             },
         )
     ),
-    model_defaults=ModelDefaultsConfig(dialect="duckdb"),
+    model_defaults=ModelDefaultsConfig(dialect="duckdb", storage_format="iceberg"),
 )
 
 
 # Due to a 3.7 mypy bug we ignore. Can remove once 3.7 support is dropped.
 airflow_config_docker = Config(  # type: ignore
     default_scheduler=AirflowSchedulerConfig(airflow_url="http://airflow-webserver:8080/"),
     gateways=GatewayConfig(connection=SparkConnectionConfig()),
-    model_defaults=ModelDefaultsConfig(dialect="duckdb"),
+    model_defaults=ModelDefaultsConfig(dialect="duckdb", storage_format="iceberg"),
 )
 
 # A DuckDB config with a physical schema map.