Spark as a Lakehouse Engine

Spark as a Lakehouse Engine, Revisited

On the lakehouse, Spark 129 's strength is writing: Iceberg 129 's procedures, write modes and MERGE INTO variants appear there first. A typical job builds a gold table, here booknest.daily_sales (one row per day and channel), which Data Quality and Anomalies, Contracts, Lineage check and monitor:

daily_sales.py: Spark builds the daily_sales gold table with MERGE INTOPython

from lake import spark
spark.sql("""CREATE OR REPLACE TABLE booknest.daily_sales (order_date DATE, channel STRING,
  orders BIGINT, cancelled BIGINT, gross DECIMAL(12,2), net DECIMAL(12,2))
  USING iceberg PARTITIONED BY (months(order_date))""")
MERGE = """MERGE INTO booknest.daily_sales t USING (
  SELECT to_date(o.order_ts) AS order_date, o.channel, count(*) AS orders,
    count_if(o.status = 'cancelled') AS cancelled, sum(o.total) AS net,
    sum(l.gross) FILTER (WHERE o.status <> 'cancelled') AS gross
  FROM booknest.orders o JOIN (SELECT order_id, sum(qty * unit_price) AS gross
    FROM booknest.order_items GROUP BY order_id) l ON l.order_id = o.order_id GROUP BY 1, 2) s
ON t.order_date = s.order_date AND t.channel = s.channel
WHEN MATCHED AND (t.orders <> s.orders OR t.net <> s.net) THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *"""
for run in (1, 2):                                  # the second run changes no values
    spark.sql(MERGE)
    s = spark.table("booknest.daily_sales.snapshots").orderBy("committed_at").collect()[-1]
    print(f"run {run}: {s.operation}, {s.summary['added-data-files']} files written")
print(spark.sql("SELECT count(*), sum(orders), sum(gross) FROM booknest.daily_sales").first())

Output

run 1: append, 18 files written
run 2: overwrite, 18 files written
Row(count(1)=1638, sum(orders)=100000, sum(gross)=Decimal('3303427.30'))

The table reconciles with the baseline (100,000 orders, 3,303,427.30), and the second run shows a pitfall: no row qualified for UPDATE, yet copy-on-write rewrote all 18 files because each held a matched row. Restrict the source to changed days, or use merge-on-read. The job took 34 seconds, a third of it starting the session.

MENU