Using Spark on Dataproc & Apache Iceberg to Build an Open Lakehouse

Raw Parquet lakes on object storage solved cheap, decoupled scalability—but left data engineers stitching together fragile directory-rename hacks whenever workloads needed row-level MERGE upserts, GDPR deletes, atomic schema evolution, or snapshot isolation.

Apache Iceberg brings full warehouse-grade ACID table semantics directly onto open Parquet files in Google Cloud Storage (GCS)—without locking your data into a single compute engine. Better yet, you can write and compact Iceberg tables at scale using Apache Spark on Google Cloud Dataproc while querying those exact same tables seamlessly from BigQuery via BigLake!

1. Launching Spark on Dataproc with Apache Iceberg

On Dataproc 2.1+ (Spark 3.3+), enable the Iceberg Spark runtime JAR and register a catalog pointing to your GCS warehouse bucket (or BigLake Metastore for multi-engine catalog governance):

gcloud dataproc jobs submit pyspark my_iceberg_job.py \
  --cluster=lakehouse-dataproc \
  --region=us-central1 \
  --jars=gs://spark-lib/iceberg/iceberg-spark-runtime-3.3_2.12-1.4.2.jar \
  --properties="^#^spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions#spark.sql.catalog.lakehouse=org.apache.iceberg.spark.SparkCatalog#spark.sql.catalog.lakehouse.type=hadoop#spark.sql.catalog.lakehouse.warehouse=gs://my-open-lakehouse-bucket/warehouse"

Tip: Need help sizing executor cores, memory, and overhead for heavy Iceberg compactions? Plug your cluster specs into my Spark Config Calculator.

2. Creating Tables with Hidden Partitioning

In legacy Hive tables, partitioning by date forced analysts to filter on a synthetic string column (event_date = '2023-12-01')—and a filter on the raw timestamp (event_ts >= TIMESTAMP '2023-12-01 00:00:00') accidentally triggered a full-table scan. Iceberg’s hidden partitioning tracks partition transforms in metadata so queries filter naturally on event_ts while still pruning files automatically:

CREATE TABLE IF NOT EXISTS lakehouse.sales.orders (
  order_id BIGINT,
  customer_id STRING,
  order_status STRING,
  amount DECIMAL(18, 2),
  event_ts TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(event_ts), bucket(16, customer_id))
TBLPROPERTIES (
  'format-version' = '2',
  'write.parquet.compression-codec' = 'zstd'
);

3. ACID Upserts with MERGE INTO & Schema Evolution

With Iceberg Format v2 enabled, CDC pipelines can apply row-level updates, deletes, and inserts atomically in a single Spark SQL statement:

MERGE INTO lakehouse.sales.orders AS target
USING staging_order_cdc AS source
ON target.order_id = source.order_id
WHEN MATCHED AND source.op = 'D' THEN DELETE
WHEN MATCHED THEN UPDATE SET
  target.order_status = source.order_status,
  target.amount = source.amount,
  target.event_ts = source.event_ts
WHEN NOT MATCHED THEN INSERT (
  order_id, customer_id, order_status, amount, event_ts
) VALUES (
  source.order_id, source.customer_id, source.order_status, source.amount, source.event_ts
);

-- Zero-rewrite schema evolution (tracked by column ID, never position/name):
ALTER TABLE lakehouse.sales.orders ADD COLUMN payment_channel STRING;

4. Snapshot Time Travel & Table Maintenance

Every commit creates an immutable metadata snapshot—making point-in-time reproducibility and instant rollbacks trivial, provided you schedule periodic small-file compaction and snapshot expiration:

-- Inspect commit history
SELECT snapshot_id, committed_at, operation, summary
FROM lakehouse.sales.orders.snapshots;

-- Query table state as of a historical timestamp
SELECT COUNT(*), SUM(amount)
FROM lakehouse.sales.orders TIMESTAMP AS OF '2023-12-05 12:00:00';

-- Compact small files & expire old snapshots to keep GCS reads fast
CALL lakehouse.system.rewrite_data_files(table => 'sales.orders');
CALL lakehouse.system.expire_snapshots(
  table => 'sales.orders',
  older_than => TIMESTAMP '2023-11-29 00:00:00'
);

Originally published in the Google Cloud Community on Medium: Using Spark on Dataproc & Apache Iceberg to Build an Open Lakehouse.