How to Build an Open Lakehouse on Your Laptop

mkdir open-lakehouse && cd open-lakehouse
python3 -m venv .venv
source .venv/bin/activate
pip install pyspark-client==4.2.0 pandas==2.3.3 boto3==1.43.93 kafka-python==2.3.0 pyiceberg==0.12.0 duckdb==1.5.6 mlflow-skinny==3.16.1

open-lakehouse/
├── .venv/                      the Python environment you just made
├── compose.yaml                every service, added layer by layer
├── s3.json                     SeaweedFS access keys
├── server.properties           Unity Catalog settings
├── spark-conf/
│   └── spark-defaults.conf     Spark settings, mounted into the Spark containers
├── pipeline/
│   ├── spark-pipeline.yml      the declarative pipeline's spec
│   └── transformations/
│       ├── bronze.py
│       ├── silver.sql
│       └── gold.sql
├── dags/
│   ├── medallion.py            the Airflow DAG that runs the pipeline
│   └── maintenance.py          the Airflow DAG that compacts and vacuums
├── common.py                   shared endpoints, imported by every script below
├── create_bucket.py
├── hello_spark.py
├── generate_orders.py
├── load_orders.py
├── register_orders.py
├── enable_uniform.py
├── read_iceberg.py
├── prepare_medallion.py
├── read_gold.py
├── stream_clean.py
├── stream_totals.py
├── check_stream.py
└── log_run.py

"""Endpoints and clients that every script in this tutorial shares."""

import json
import os
import urllib.request

import boto3
from pyspark.sql import SparkSession

SPARK_REMOTE = os.environ.get("SPARK_REMOTE", "sc://localhost:15002")
S3_ENDPOINT = os.environ.get("S3_ENDPOINT", "http://localhost:8333")
UC_URL = os.environ.get("UC_URL", "http://localhost:8081")
KAFKA_BOOTSTRAP = os.environ.get("KAFKA_BOOTSTRAP", "localhost:9092")
MLFLOW_TRACKING_URI = os.environ.get("MLFLOW_TRACKING_URI", "http://localhost:5000")
S3_KEY, S3_SECRET = "lakehouse", "lakehouse-secret"


def spark():
    """A Spark Connect session: a thin client, the work runs in the cluster."""
    return SparkSession.builder.remote(SPARK_REMOTE).getOrCreate()


def s3():
    """An S3 client for SeaweedFS."""
    return boto3.client(
        "s3",
        endpoint_url=S3_ENDPOINT,
        aws_access_key_id=S3_KEY,
        aws_secret_access_key=S3_SECRET,
        region_name="us-east-1",
    )


def uc(method, path, body=None):
    """Call the Unity Catalog REST API and return the JSON reply."""
    request = urllib.request.Request(
        f"{UC_URL}/api/2.1/unity-catalog{path}",
        data=json.dumps(body).encode() if body else None,
        headers={"Content-Type": "application/json"},
        method=method,
    )
    with urllib.request.urlopen(request) as reply:
        return json.load(reply)

{
  "identities": [
    {
      "name": "lakehouse",
      "credentials": [{ "accessKey": "lakehouse", "secretKey": "lakehouse-secret" }],
      "actions": ["Admin", "Read", "Write", "List", "Tagging"]
    }
  ]
}

services:
  seaweedfs:
    image: chrislusf/seaweedfs:3.80
    entrypoint: weed
    command: server -dir=/data -s3 -s3.config=/etc/seaweedfs/s3.json -volume.max=0 -master.volumeSizeLimitMB=1024
    ports:
      - "${S3_PORT:-8333}:8333"
    volumes:
      - ./s3.json:/etc/seaweedfs/s3.json:ro
      - seaweedfs-data:/data
    healthcheck:
      test: wget -qO /dev/null http://127.0.0.1:8333/status
      interval: 5s
      retries: 30

volumes:
  seaweedfs-data:

docker compose up -d --wait

"""Create the `lakehouse` bucket, write one object and list it back."""

from common import s3

client = s3()
existing = [b["Name"] for b in client.list_buckets()["Buckets"]]
if "lakehouse" not in existing:
    client.create_bucket(Bucket="lakehouse")

client.put_object(Bucket="lakehouse", Key="hello.txt", Body=b"hello, lakehouse\n")
for obj in client.list_objects_v2(Bucket="lakehouse")["Contents"]:
    print(obj["Key"], obj["Size"], "bytes")

python create_bucket.py

hello.txt 17 bytes

mkdir spark-conf

# Delta Lake
spark.sql.extensions                             io.delta.sql.DeltaSparkSessionExtension
spark.sql.catalog.spark_catalog                  org.apache.spark.sql.delta.catalog.DeltaCatalog

# SeaweedFS, through Hadoop's S3A connector, for both s3:// and s3a:// paths
spark.hadoop.fs.s3a.endpoint                     http://seaweedfs:8333
spark.hadoop.fs.s3a.access.key                   lakehouse
spark.hadoop.fs.s3a.secret.key                   lakehouse-secret
spark.hadoop.fs.s3a.path.style.access            true
spark.hadoop.fs.s3a.connection.ssl.enabled       false
spark.hadoop.fs.s3.impl                          org.apache.hadoop.fs.s3a.S3AFileSystem
spark.hadoop.fs.AbstractFileSystem.s3.impl       org.apache.hadoop.fs.s3a.S3A

# Resources: a cap, so other applications still get cores, and idle executors go back
spark.driver.memory                              2g
spark.executor.memory                            1g
spark.executor.cores                             2
spark.cores.max                                  4
spark.dynamicAllocation.enabled                  true
spark.dynamicAllocation.shuffleTracking.enabled  true
spark.dynamicAllocation.executorIdleTimeout      60s
spark.dynamicAllocation.shuffleTracking.timeout  120s
spark.dynamicAllocation.cachedExecutorIdleTimeout 300s

# Where --packages keeps the JARs it downloads (a volume, so it happens once)
spark.jars.ivy                                   /opt/spark/work-dir/ivy

  spark-master:
    image: apache/spark:4.2.0
    hostname: spark-master
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.master.Master --host spark-master
    ports:
      - "${SPARK_UI_PORT:-8080}:8080"
    healthcheck:
      test: curl -sf http://localhost:8080 > /dev/null
      interval: 5s
      retries: 30

  spark-worker:
    image: apache/spark:4.2.0
    hostname: spark-worker
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.worker.Worker --host spark-worker spark://spark-master:7077
    environment:
      SPARK_WORKER_CORES: "6"
      SPARK_WORKER_MEMORY: 4g
    volumes:
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy

  spark-connect:
    image: apache/spark:4.2.0
    hostname: spark-connect
    command: >
      /opt/spark/sbin/start-connect-server.sh
      --master spark://spark-master:7077
      --packages io.delta:delta-spark_4.2_2.13:4.4.0,io.delta:delta-iceberg_2.13:4.4.0,io.unitycatalog:unitycatalog-spark_4.2_2.13:0.6.0,org.apache.hadoop:hadoop-aws:3.5.0,org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0
      --exclude-packages io.delta:delta-spark_4.1_2.13
    environment:
      SPARK_NO_DAEMONIZE: "1"
    ports:
      - "${SPARK_CONNECT_PORT:-15002}:15002"
    volumes:
      - ./spark-conf:/opt/spark/conf
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "bash", "-c", "echo > /dev/tcp/localhost/15002"]
      interval: 5s
      start_period: 15m

  spark-work:

services:
  seaweedfs:
    image: chrislusf/seaweedfs:3.80
    entrypoint: weed
    command: server -dir=/data -s3 -s3.config=/etc/seaweedfs/s3.json -volume.max=0 -master.volumeSizeLimitMB=1024
    ports:
      - "${S3_PORT:-8333}:8333"
    volumes:
      - ./s3.json:/etc/seaweedfs/s3.json:ro
      - seaweedfs-data:/data
    healthcheck:
      test: wget -qO /dev/null http://127.0.0.1:8333/status
      interval: 5s
      retries: 30

  spark-master:
    image: apache/spark:4.2.0
    hostname: spark-master
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.master.Master --host spark-master
    ports:
      - "${SPARK_UI_PORT:-8080}:8080"
    healthcheck:
      test: curl -sf http://localhost:8080 > /dev/null
      interval: 5s
      retries: 30

  spark-worker:
    image: apache/spark:4.2.0
    hostname: spark-worker
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.worker.Worker --host spark-worker spark://spark-master:7077
    environment:
      SPARK_WORKER_CORES: "6"
      SPARK_WORKER_MEMORY: 4g
    volumes:
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy

  spark-connect:
    image: apache/spark:4.2.0
    hostname: spark-connect
    command: >
      /opt/spark/sbin/start-connect-server.sh
      --master spark://spark-master:7077
      --packages io.delta:delta-spark_4.2_2.13:4.4.0,io.delta:delta-iceberg_2.13:4.4.0,io.unitycatalog:unitycatalog-spark_4.2_2.13:0.6.0,org.apache.hadoop:hadoop-aws:3.5.0,org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0
      --exclude-packages io.delta:delta-spark_4.1_2.13
    environment:
      SPARK_NO_DAEMONIZE: "1"
    ports:
      - "${SPARK_CONNECT_PORT:-15002}:15002"
    volumes:
      - ./spark-conf:/opt/spark/conf
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "bash", "-c", "echo > /dev/tcp/localhost/15002"]
      interval: 5s
      start_period: 15m

volumes:
  seaweedfs-data:
  spark-work:

docker compose up -d --wait

"""Send one query to the cluster through Spark Connect."""

from common import SPARK_REMOTE, spark

session = spark()
print(f"Connected to Spark {session.version} at {SPARK_REMOTE}")
session.range(1_000_000).selectExpr("sum(id) AS total").show()

python hello_spark.py

Connected to Spark 4.2.0 at sc://localhost:15002
+------------+
|       total|
+------------+
|499999500000|
+------------+

"""Ghost-kitchen orders: a week of history for the bucket, or a live stream for Kafka.

    python generate_orders.py files    # 7 days of events -> s3://lakehouse/landing/
    python generate_orders.py stream   # one day of events, live -> Kafka topic `orders`
"""

import io
import json
import random
import sys
import time
from datetime import datetime, timedelta, timezone

import pandas as pd
from common import KAFKA_BOOTSTRAP, s3

LOCATIONS = {1: "Oakland", 2: "Berkeley", 3: "San Francisco", 4: "San Jose"}
BRANDS = {"Pizza Planet": 16.0, "Wok This Way": 21.0, "Taco Loco": 12.5,
          "Curry House": 23.0, "Sushi Express": 29.0}
rng = random.Random(7)


def order(ts):
    """One order's events: created, ready in the kitchen, delivered."""
    order_id = f"{rng.getrandbits(40):010x}"
    location = rng.choice(list(LOCATIONS))
    brand, items = rng.choice(list(BRANDS)), rng.randint(1, 4)
    total = round(BRANDS[brand] * items * rng.uniform(0.9, 1.2), 2)
    steps = [("order_created", 0), ("order_ready", rng.randint(8, 20)),
             ("delivered", rng.randint(25, 50))]
    return [{
        "event_id": f"{order_id}-{n}",
        "event_type": kind,
        "ts": (ts + timedelta(minutes=minutes)).isoformat(timespec="seconds"),
        "location_id": location,
        "order_id": order_id,
        "body": json.dumps({"brand": brand, "items": items, "total": total}),
    } for n, (kind, minutes) in enumerate(steps)]


def messy(events):
    """Real feeds misbehave: a few events lose their location, some arrive twice."""
    out = []
    for event in events:
        if rng.random() < 0.01:
            event["location_id"] = None
        out.append(event)
        if rng.random() < 0.03:
            out.append(dict(event))
    return out


def week():
    start = datetime(2026, 9, 1)
    for day in range(7):
        events = []
        for _ in range(rng.randint(1200, 1800)):
            events += order(start + timedelta(days=day, seconds=rng.randint(36000, 79200)))
        yield (start + timedelta(days=day)).date(), messy(events)


def put_parquet(client, key, frame):
    buffer = io.BytesIO()
    frame.to_parquet(buffer, index=False)
    client.put_object(Bucket="lakehouse", Key=key, Body=buffer.getvalue())
    print(f"{key}  {len(frame):,} rows")


def write_files():
    client = s3()
    for day, events in week():
        frame = pd.DataFrame(events).astype({"location_id": "Int64"})  # keep the gaps as nulls
        put_parquet(client, f"landing/orders/{day}.parquet", frame)
    cities = pd.DataFrame({"location_id": list(LOCATIONS), "city": list(LOCATIONS.values())})
    put_parquet(client, "landing/locations/locations.parquet", cities)


def stream(per_second=50):
    from kafka import KafkaProducer

    producer = KafkaProducer(bootstrap_servers=KAFKA_BOOTSTRAP, key_serializer=str.encode,
                             value_serializer=lambda v: json.dumps(v).encode())
    rng.seed()  # new order ids on every run
    _, events = next(week())
    events.sort(key=lambda e: e["ts"])
    stamps = {}
    for n, event in enumerate(events, 1):
        if event["event_id"] not in stamps:  # a resent event keeps its first timestamp
            late = rng.random() < 0.02  # and a few arrive three minutes late
            now = datetime.now(timezone.utc) - timedelta(minutes=3 if late else 0)
            stamps[event["event_id"]] = now.isoformat(timespec="milliseconds")
        producer.send("orders", key=event["order_id"],
                      value={**event, "ts": stamps[event["event_id"]]})
        if n % 1000 == 0:
            print(f"sent {n:,} events")
        time.sleep(1 / per_second)
    producer.flush()
    print(f"sent {len(events):,} events to the topic orders")


if __name__ == "__main__":
    {"files": write_files, "stream": stream}[sys.argv[1]]()

python generate_orders.py files

landing/orders/2026-09-01.parquet  4,720 rows
landing/orders/2026-09-02.parquet  4,382 rows
landing/orders/2026-09-03.parquet  5,479 rows
landing/orders/2026-09-04.parquet  4,508 rows
landing/orders/2026-09-05.parquet  5,194 rows
landing/orders/2026-09-06.parquet  4,962 rows
landing/orders/2026-09-07.parquet  4,776 rows
landing/locations/locations.parquet  4 rows

"""Write the week's events as a Delta table in two commits, then time-travel."""

from common import s3, spark
from pyspark.sql import functions as f

TABLE_PATH = "s3://lakehouse/tables/orders_raw"

session = spark()
events = session.read.parquet("s3://lakehouse/landing/orders/").withColumn(
    "event_date", f.to_date("ts")
)

# Commit 0: the first day. Commit 1: the other six.
events.where("event_date = '2026-09-01'").write.format("delta").save(TABLE_PATH)
events.where("event_date > '2026-09-01'").write.format("delta").mode("append").save(TABLE_PATH)

session.sql(f"DESCRIBE HISTORY delta.`{TABLE_PATH}`").select(
    "version", "operation", "operationMetrics.numOutputRows"
).show()

now = session.read.format("delta").load(TABLE_PATH).count()
then = session.read.format("delta").option("versionAsOf", 0).load(TABLE_PATH).count()
print(f"rows now: {now:,}   rows at version 0: {then:,}\n")

# Underneath, it's files in the bucket: Parquet data plus the _delta_log.
for obj in s3().list_objects_v2(Bucket="lakehouse", Prefix="tables/orders_raw/")["Contents"]:
    print(obj["Key"])

python load_orders.py

+-------+---------+-------------+
|version|operation|numOutputRows|
+-------+---------+-------------+
|      1|    WRITE|        29301|
|      0|    WRITE|         4720|
+-------+---------+-------------+

rows now: 34,021   rows at version 0: 4,720

tables/orders_raw/_delta_log/00000000000000000000.crc
tables/orders_raw/_delta_log/00000000000000000000.json
tables/orders_raw/_delta_log/00000000000000000001.crc
tables/orders_raw/_delta_log/00000000000000000001.json
tables/orders_raw/_delta_log/_staged_commits/
tables/orders_raw/part-00000-7f3f327f-b1e1-4f75-99de-ccfbc82c1b65-c000.snappy.parquet
tables/orders_raw/part-00000-92a94a61-efd3-4a85-924d-e3481bde0ed6-c000.snappy.parquet
tables/orders_raw/part-00001-9db8f5d3-76b4-4a7b-972b-b3b51c7f29e3-c000.snappy.parquet
tables/orders_raw/part-00001-fd4c4fe3-30ec-43f1-a3b6-3b5128241edd-c000.snappy.parquet

server.env=dev
server.authorization=disable

# The bucket Unity Catalog hands out credentials for, and the key pair it uses
s3.bucketPath.0=s3://lakehouse
s3.region.0=us-east-1
s3.accessKey.0=lakehouse
s3.secretKey.0=lakehouse-secret
s3.sessionToken.0=unused
s3.endpoint.0=http://seaweedfs:8333

  unity-catalog:
    image: unitycatalog/unitycatalog:v0.6.0
    ports:
      - "${UC_PORT:-8081}:8080"
    volumes:
      - ./server.properties:/home/unitycatalog/etc/conf/server.properties:ro
      - uc-data:/home/unitycatalog/etc/db
    healthcheck:
      test: wget -qO /dev/null http://127.0.0.1:8080/api/2.1/unity-catalog/catalogs
      interval: 5s
      retries: 30

  uc-data:

# Unity Catalog: Spark's catalog `lakehouse` is the Unity Catalog catalog `lakehouse`
spark.sql.catalog.lakehouse                      io.unitycatalog.spark.UCSingleCatalog
spark.sql.catalog.lakehouse.uri                  http://unity-catalog:8080
spark.sql.catalog.lakehouse.token                unused

docker compose up -d --wait
docker compose up -d --wait --force-recreate spark-connect

"""Create the `lakehouse` catalog, register orders_raw in it, and query it by name."""

from common import spark, uc

if "lakehouse" not in [c["name"] for c in uc("GET", "/catalogs")["catalogs"]]:
    uc("POST", "/catalogs", {"name": "lakehouse", "storage_root": "s3://lakehouse/managed"})

session = spark()
session.sql("CREATE SCHEMA IF NOT EXISTS lakehouse.tutorial")
session.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.tutorial.orders_raw
    USING delta LOCATION 's3://lakehouse/tables/orders_raw'
""")
session.sql("""
    SELECT event_date, count(*) AS events
    FROM lakehouse.tutorial.orders_raw
    GROUP BY event_date ORDER BY event_date
""").show()

# Ask the catalog itself, over its REST API.
for table in uc("GET", "/tables?catalog_name=lakehouse&schema_name=tutorial")["tables"]:
    print(table["name"], table["table_type"], table["data_source_format"], table["storage_location"])

python register_orders.py

+----------+------+
|event_date|events|
+----------+------+
|2026-09-01|  4720|
|2026-09-02|  4382|
|2026-09-03|  5479|
|2026-09-04|  4508|
|2026-09-05|  5194|
|2026-09-06|  4962|
|2026-09-07|  4776|
+----------+------+

orders_raw EXTERNAL DELTA s3://lakehouse/tables/orders_raw

"""Turn on UniForm: orders_raw stays a Delta table and also gets Iceberg metadata."""

from common import s3, spark

spark().sql("""
    ALTER TABLE lakehouse.tutorial.orders_raw SET TBLPROPERTIES (
        'delta.universalFormat.enabledFormats' = 'iceberg',
        'delta.enableIcebergCompatV2' = 'true',
        'delta.columnMapping.mode' = 'name'
    )
""")

for obj in s3().list_objects_v2(Bucket="lakehouse", Prefix="tables/orders_raw/metadata/")["Contents"]:
    print(obj["Key"])

python enable_uniform.py

tables/orders_raw/metadata/00000-01bfacd6-e573-475c-917f-34bc3a647658.metadata.json
tables/orders_raw/metadata/ec2026c7-0ade-4f27-a333-598ff9fa2096-m0.avro
tables/orders_raw/metadata/snap-2861189516262794144-1-ec2026c7-0ade-4f27-a333-598ff9fa2096.avro

"""Read orders_raw as an Iceberg table with PyIceberg and DuckDB. No Spark involved."""

import duckdb
from common import S3_ENDPOINT, S3_KEY, S3_SECRET, s3
from pyiceberg.table import StaticTable

# The table's current Iceberg metadata is the newest metadata.json UniForm wrote.
listing = s3().list_objects_v2(Bucket="lakehouse", Prefix="tables/orders_raw/metadata/")
newest = max(
    (obj for obj in listing["Contents"] if obj["Key"].endswith(".metadata.json")),
    key=lambda obj: obj["LastModified"],
)
location = f"s3://lakehouse/{newest['Key']}"
print(f"Reading {location}\n")

table = StaticTable.from_metadata(location, properties={
    "s3.endpoint": S3_ENDPOINT,
    "s3.access-key-id": S3_KEY,
    "s3.secret-access-key": S3_SECRET,
    "s3.region": "us-east-1",
})
rows = table.scan(selected_fields=("event_id",)).to_arrow().num_rows
print(f"PyIceberg: {rows:,} rows (Iceberg v{table.format_version}, "
      f"converted from Delta version {table.properties['delta-version']})\n")

duck = duckdb.connect()
duck.sql("INSTALL iceberg; LOAD iceberg; INSTALL httpfs; LOAD httpfs;")
duck.sql(f"""
    CREATE SECRET (TYPE s3, KEY_ID '{S3_KEY}', SECRET '{S3_SECRET}', REGION 'us-east-1',
                   ENDPOINT '{S3_ENDPOINT.split("://")[1]}', URL_STYLE 'path', USE_SSL false)
""")
duck.sql(f"""
    SELECT event_date, count(*) AS events
    FROM iceberg_scan('{location}')
    GROUP BY event_date ORDER BY event_date
""").show()

python read_iceberg.py

Reading s3://lakehouse/tables/orders_raw/metadata/00000-01bfacd6-e573-475c-917f-34bc3a647658.metadata.json

PyIceberg: 34,021 rows (Iceberg v2, converted from Delta version 2)

┌────────────┬────────┐
│ event_date │ events │
│    date    │ int64  │
├────────────┼────────┤
│ 2026-09-01 │   4720 │
│ 2026-09-02 │   4382 │
│ 2026-09-03 │   5479 │
│ 2026-09-04 │   4508 │
│ 2026-09-05 │   5194 │
│ 2026-09-06 │   4962 │
│ 2026-09-07 │   4776 │
└────────────┴────────┘

mkdir -p pipeline/transformations

name: medallion
catalog: lakehouse
database: medallion
storage: s3://lakehouse/pipelines/medallion
libraries:
  - glob:
      include: transformations/**

"""Bronze: the landing files, as they arrived."""

from pyspark import pipelines as dp
from pyspark.sql import SparkSession

spark = SparkSession.active()

# Unity Catalog manages these tables: it picks their location under the
# catalog's storage root. These properties mark a Delta table as catalog-managed.
MANAGED = {
    "delta.feature.catalogManaged": "supported",
    "delta.checkpoint.writeStatsAsJson": "true",
    "delta.checkpoint.writeStatsAsStruct": "true",
}


@dp.materialized_view(format="delta", table_properties=MANAGED)
def orders_bronze():
    return spark.read.parquet("s3://lakehouse/landing/orders/")


@dp.materialized_view(format="delta", table_properties=MANAGED)
def locations():
    return spark.read.parquet("s3://lakehouse/landing/locations/")

-- Silver: each event once, with a known location, the order details parsed
-- out of the JSON body, and the city attached.
CREATE MATERIALIZED VIEW orders_silver
USING delta
TBLPROPERTIES ('delta.feature.catalogManaged' = 'supported',
               'delta.checkpoint.writeStatsAsJson' = 'true',
               'delta.checkpoint.writeStatsAsStruct' = 'true')
AS SELECT DISTINCT
  o.event_id,
  o.event_type,
  to_timestamp(o.ts) AS event_time,
  o.order_id,
  l.city,
  get_json_object(o.body, '$.brand') AS brand,
  CAST(get_json_object(o.body, '$.total') AS DOUBLE) AS total
FROM orders_bronze o
JOIN locations l ON o.location_id = l.location_id;

-- Gold: the numbers people ask for, built from silver.
CREATE MATERIALIZED VIEW revenue_by_brand
USING delta
TBLPROPERTIES ('delta.feature.catalogManaged' = 'supported',
               'delta.checkpoint.writeStatsAsJson' = 'true',
               'delta.checkpoint.writeStatsAsStruct' = 'true')
AS SELECT
  brand,
  count(*) AS orders,
  round(sum(total), 2) AS revenue,
  round(avg(total), 2) AS avg_order
FROM orders_silver
WHERE event_type = 'order_created'
GROUP BY brand;

CREATE MATERIALIZED VIEW revenue_by_city_day
USING delta
TBLPROPERTIES ('delta.feature.catalogManaged' = 'supported',
               'delta.checkpoint.writeStatsAsJson' = 'true',
               'delta.checkpoint.writeStatsAsStruct' = 'true')
AS SELECT
  city,
  to_date(event_time) AS day,
  count(*) AS orders,
  round(sum(total), 2) AS revenue
FROM orders_silver
WHERE event_type = 'order_created'
GROUP BY city, to_date(event_time);

"""Get lakehouse.medallion ready for a pipeline run: create it, or clear the last run.

SDP can't replace a table that already exists in Unity Catalog, so a rerun of the
pipeline starts by dropping the tables the previous run created.
"""

from common import spark

session = spark()
session.sql("CREATE SCHEMA IF NOT EXISTS lakehouse.medallion")
for row in session.sql("SHOW TABLES IN lakehouse.medallion").collect():
    session.sql(f"DROP TABLE lakehouse.medallion.{row.tableName}")
    print(f"dropped lakehouse.medallion.{row.tableName}")
print("lakehouse.medallion is ready")

python prepare_medallion.py
SPARK_REMOTE=sc://localhost:15002 python -m pyspark.pipelines.cli dry-run --spec pipeline/spark-pipeline.yml
SPARK_REMOTE=sc://localhost:15002 python -m pyspark.pipelines.cli run --spec pipeline/spark-pipeline.yml

2026-10-01 14:36:58: Loading definitions. Root directory: '.../open-lakehouse/pipeline'.
2026-10-01 14:36:58: Found 3 files matching glob 'transformations/**/*'
...
2026-10-01 21:37:09: Flow lakehouse.medallion.orders_bronze has COMPLETED.
2026-10-01 21:37:09: Flow lakehouse.medallion.locations has COMPLETED.
2026-10-01 21:37:15: Flow lakehouse.medallion.orders_silver has COMPLETED.
2026-10-01 21:37:19: Flow lakehouse.medallion.revenue_by_city_day has COMPLETED.
2026-10-01 21:37:19: Flow lakehouse.medallion.revenue_by_brand has COMPLETED.
2026-10-01 21:37:21: Run is COMPLETED.

"""Read a gold table by name, and list what the pipeline put in the catalog."""

from common import spark, uc

spark().sql("""
    SELECT * FROM lakehouse.medallion.revenue_by_brand ORDER BY revenue DESC
""").show()

for table in uc("GET", "/tables?catalog_name=lakehouse&schema_name=medallion")["tables"]:
    print(table["name"], table["table_type"], table["data_source_format"])

python read_gold.py

+-------------+------+---------+---------+
|        brand|orders|  revenue|avg_order|
+-------------+------+---------+---------+
|Sushi Express|  2140|160332.61|    74.92|
|  Curry House|  2219|133432.76|    60.13|
| Wok This Way|  2152|118668.13|    55.14|
| Pizza Planet|  2233| 93402.24|    41.83|
|    Taco Loco|  2168| 71548.53|     33.0|
+-------------+------+---------+---------+

locations MANAGED DELTA
orders_bronze MANAGED DELTA
orders_silver MANAGED DELTA
revenue_by_brand MANAGED DELTA
revenue_by_city_day MANAGED DELTA

  kafka:
    image: apache/kafka:4.2.0
    ports:
      - "${KAFKA_PORT:-9092}:9094"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: INTERNAL://:9092,CONTROLLER://:9093,EXTERNAL://:9094
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://localhost:${KAFKA_PORT:-9092}
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
    healthcheck:
      test: /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
      interval: 10s
      retries: 30

docker compose up -d --wait

docker compose exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --topic orders --partitions 2
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --topic orders-clean --partitions 2

Created topic orders.
Created topic orders-clean.

"""Real-Time Mode: clean each order event from Kafka into another topic as it arrives."""

from common import spark
from pyspark.sql import functions as f

EVENT = "event_id STRING, event_type STRING, ts STRING, location_id INT, order_id STRING, body STRING"

session = spark()
orders = (
    session.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")  # Kafka as the cluster sees it
    .option("subscribe", "orders")
    .load()
    .select(f.from_json(f.col("value").cast("string"), EVENT).alias("e"), "timestamp")
)

clean = orders.where("e.event_type = 'order_created' AND e.location_id IS NOT NULL").select(
    f.col("e.order_id").alias("key"),
    f.to_json(f.struct(
        "e.event_id",
        "e.location_id",
        f.get_json_object("e.body", "$.brand").alias("brand"),
        f.get_json_object("e.body", "$.total").cast("double").alias("total"),
        f.unix_millis("timestamp").alias("received_ms"),  # when it reached Kafka
    )).alias("value"),
)

query = (
    clean.writeStream.queryName("orders_clean")
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("topic", "orders-clean")
    .option("checkpointLocation", "/opt/spark/work-dir/checkpoints/orders_clean")
    .outputMode("update")  # the output mode Real-Time Mode supports
    .trigger(realTime="5 minutes")  # how often to checkpoint, not a latency target
    .start()
)
print("orders -> orders-clean, in Real-Time Mode. Ctrl+C to stop.")
try:
    query.awaitTermination()
except KeyboardInterrupt:
    query.stop()

"""Micro-batch: deduplicate orders and keep running totals per location, in Delta."""

from common import spark
from pyspark.sql import functions as f

EVENT = "event_id STRING, event_type STRING, ts STRING, location_id INT, order_id STRING, body STRING"

session = spark()
session.conf.set("spark.sql.shuffle.partitions", "4")
session.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.tutorial.live_totals (
        location_id INT, orders BIGINT, revenue DOUBLE, last_order_at TIMESTAMP
    ) USING delta LOCATION 's3://lakehouse/tables/live_totals'
""")

totals = (
    session.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "orders")
    .load()
    .select(f.from_json(f.col("value").cast("string"), EVENT).alias("e"))
    .select("e.*", f.to_timestamp("e.ts").alias("event_time"))
    .where("event_type = 'order_created' AND location_id IS NOT NULL")
    .withWatermark("event_time", "10 minutes")  # how long to remember an event id
    .dropDuplicates(["event_id", "event_time"])
    .groupBy("location_id")
    .agg(
        f.count("*").alias("orders"),
        f.round(f.sum(f.get_json_object("body", "$.total").cast("double")), 2).alias("revenue"),
        f.max("event_time").alias("last_order_at"),
    )
)

query = (
    totals.writeStream.queryName("live_totals")
    .format("delta")
    .outputMode("complete")  # each batch rewrites the four rows with the current totals
    .option("checkpointLocation", "/opt/spark/work-dir/checkpoints/live_totals")
    .trigger(processingTime="1 second")
    .toTable("lakehouse.tutorial.live_totals")
)
print("orders -> lakehouse.tutorial.live_totals, every second. Ctrl+C to stop.")
try:
    query.awaitTermination()
except KeyboardInterrupt:
    query.stop()

python stream_clean.py

python stream_totals.py

python generate_orders.py stream

sent 1,000 events
...
sent 4,625 events to the topic orders

docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic orders-clean --from-beginning --max-messages 3

{"event_id":"4de57cb207-0","location_id":4,"brand":"Curry House","total":23.98,"received_ms":1790890668647}
{"event_id":"904638cd9b-0","location_id":2,"brand":"Wok This Way","total":23.63,"received_ms":1790890668769}
{"event_id":"bdbede7a39-0","location_id":3,"brand":"Pizza Planet","total":33.93,"received_ms":1790890668890}
Processed a total of 3 messages

"""Check both streaming queries against what the generator sent."""

from common import spark
from pyspark.sql import functions as f

EVENT = "event_id STRING, event_type STRING, ts STRING, location_id INT, order_id STRING, body STRING"

session = spark()


def topic(name):
    """Everything in a Kafka topic so far, read once as a batch."""
    return (
        session.read.format("kafka")
        .option("kafka.bootstrap.servers", "kafka:9092")
        .option("subscribe", name)
        .option("startingOffsets", "earliest")
        .load()
    )


sent = (
    topic("orders")
    .select(f.from_json(f.col("value").cast("string"), EVENT).alias("e"))
    .where("e.event_type = 'order_created' AND e.location_id IS NOT NULL")
    .agg(f.count("*").alias("records"), f.count_distinct("e.event_id").alias("distinct"))
    .first()
)
totals = session.table("lakehouse.tutorial.live_totals").orderBy("location_id")
totals.show(truncate=False)

print(f"order_created records sent:        {sent['records']:,}")
print(f"distinct order_created event ids:  {sent['distinct']:,}")
print(f"orders in live_totals:             {totals.agg(f.sum('orders')).first()[0]:,}")

# How long each record spent in the Real-Time Mode query, Kafka to Kafka.
spent = topic("orders-clean").select(
    (f.unix_millis("timestamp")
     - f.get_json_object(f.col("value").cast("string"), "$.received_ms").cast("long")).alias("ms")
)
p50, p99 = spent.agg(f.percentile_approx("ms", [0.5, 0.99])).first()[0]
print(f"\norders-clean: {spent.count():,} records, p50 {p50} ms, p99 {p99} ms in Spark")

python check_stream.py

+-----------+------+--------+-----------------------+
|location_id|orders|revenue |last_order_at          |
+-----------+------+--------+-----------------------+
|1          |361   |18666.46|2026-10-01 21:39:19.782|
|2          |354   |19276.01|2026-10-01 21:39:19.58 |
|3          |380   |19758.75|2026-10-01 21:39:19.802|
|4          |384   |19255.33|2026-10-01 21:39:19.883|
+-----------+------+--------+-----------------------+

order_created records sent:        1,512
distinct order_created event ids:  1,479
orders in live_totals:             1,479

orders-clean: 1,512 records, p50 1 ms, p99 3 ms in Spark

mkdir dags

"""Run the medallion pipeline every day, on the cluster, through Spark Connect."""

from datetime import datetime

from airflow.providers.apache.spark.operators.spark_pipelines import SparkPipelinesOperator
from airflow.sdk import DAG, task

with DAG("medallion", schedule="@daily", start_date=datetime(2026, 9, 1), catchup=False):

    @task
    def prepare():
        """Create the schema, or drop the last run's tables (SDP won't replace them)."""
        from pyspark.sql import SparkSession

        spark = SparkSession.builder.remote("sc://spark-connect:15002").getOrCreate()
        spark.sql("CREATE SCHEMA IF NOT EXISTS lakehouse.medallion")
        for row in spark.sql("SHOW TABLES IN lakehouse.medallion").collect():
            spark.sql(f"DROP TABLE lakehouse.medallion.{row.tableName}")

    run_pipeline = SparkPipelinesOperator(
        task_id="run_pipeline",
        pipeline_spec="/opt/airflow/pipeline/spark-pipeline.yml",
        pipeline_command="run",
        conn_id="spark_connect_default",
    )

    @task
    def check_gold():
        from pyspark.sql import SparkSession

        spark = SparkSession.builder.remote("sc://spark-connect:15002").getOrCreate()
        brands = spark.table("lakehouse.medallion.revenue_by_brand").count()
        print(f"revenue_by_brand has {brands} rows")
        if brands == 0:
            raise ValueError("the gold table is empty")

    prepare() >> run_pipeline >> check_gold()

"""Every night, compact the external Delta tables and clean up files nothing needs."""

from datetime import datetime

from airflow.sdk import DAG, task

TABLES = ["lakehouse.tutorial.orders_raw", "lakehouse.tutorial.live_totals"]

with DAG("delta_maintenance", schedule="0 3 * * *", start_date=datetime(2026, 9, 1), catchup=False):

    @task
    def optimize_and_vacuum():
        from pyspark.sql import SparkSession

        spark = SparkSession.builder.remote("sc://spark-connect:15002").getOrCreate()
        for table in TABLES:
            before = spark.sql(f"DESCRIBE DETAIL {table}").first()["numFiles"]
            spark.sql(f"OPTIMIZE {table}").collect()  # many small files into a few big ones
            spark.sql(f"VACUUM {table} RETAIN 168 HOURS").collect()  # keep a week of history
            after = spark.sql(f"DESCRIBE DETAIL {table}").first()["numFiles"]
            print(f"{table}: {before} files -> {after}")

    optimize_and_vacuum()

  airflow:
    build:
      dockerfile_inline: |
        FROM apache/airflow:3.3.2-python3.12
        RUN pip install --no-cache-dir apache-airflow==3.3.2 apache-airflow-providers-apache-spark==6.3.2 pyspark-client==4.2.0
    command: standalone
    environment:
      AIRFLOW__CORE__LOAD_EXAMPLES: "false"
      AIRFLOW__CORE__SIMPLE_AUTH_MANAGER_ALL_ADMINS: "true"
      AIRFLOW_CONN_SPARK_CONNECT_DEFAULT: '{"conn_type": "spark_connect", "host": "spark-connect", "port": 15002}'
    ports:
      - "${AIRFLOW_PORT:-8085}:8080"
    volumes:
      - airflow-data:/opt/airflow
      - ./dags:/opt/airflow/dags
      - ./pipeline:/opt/airflow/pipeline
    healthcheck:
      test: curl -sf http://localhost:8080/api/v2/monitor/health
      interval: 5s
      start_period: 5m

  airflow-data:

docker compose up -d --wait

docker compose exec airflow airflow dags unpause medallion
docker compose exec airflow airflow dags unpause delta_maintenance

docker compose exec airflow airflow dags list-runs medallion -o plain
docker compose exec airflow airflow dags list-runs delta_maintenance -o plain

dag_id     run_id                                state    run_after                  logical_date               start_date                        end_date
medallion  scheduled__2026-10-01T00:00:00+00:00  success  2026-10-01T00:00:00+00:00  2026-10-01T00:00:00+00:00  2026-10-01T21:40:07.659874+00:00  2026-10-01T21:40:49.719529+00:00
dag_id             run_id                                state    run_after                  logical_date               start_date                        end_date
delta_maintenance  scheduled__2026-10-01T03:00:00+00:00  success  2026-10-01T03:00:00+00:00  2026-10-01T03:00:00+00:00  2026-10-01T21:40:09.561521+00:00  2026-10-01T21:40:41.881096+00:00

docker compose exec airflow grep -rhoE "revenue_by_brand has [0-9]+ rows|lakehouse[.a-z_]+: [0-9]+ files -> [0-9]+" /opt/airflow/logs

lakehouse.tutorial.orders_raw: 3 files -> 1
lakehouse.tutorial.live_totals: 2 files -> 1
revenue_by_brand has 5 rows

python read_iceberg.py

Reading s3://lakehouse/tables/orders_raw/metadata/00000-47aecc6b-89d1-40f8-97eb-b8c390180aca.metadata.json

PyIceberg: 34,021 rows (Iceberg v2, converted from Delta version 3)
...

  mlflow:
    build:
      dockerfile_inline: |
        FROM ghcr.io/mlflow/mlflow:v3.16.1
        RUN pip install --no-cache-dir boto3==1.43.93
    command: >
      mlflow server --host 0.0.0.0 --port 5000 --workers 1
      --backend-store-uri sqlite:////mlflow/mlflow.db
      --artifacts-destination s3://lakehouse/mlflow
    environment:
      MLFLOW_S3_ENDPOINT_URL: http://seaweedfs:8333
      AWS_ACCESS_KEY_ID: lakehouse
      AWS_SECRET_ACCESS_KEY: lakehouse-secret
    ports:
      - "${MLFLOW_PORT:-5000}:5000"
    volumes:
      - mlflow-data:/mlflow
    healthcheck:
      test: python -c "import urllib.request; urllib.request.urlopen('http://localhost:5000/health')"
      interval: 5s
      retries: 30

  mlflow-data:

docker compose up -d --wait

"""Log an MLflow run from a PySpark job, then read it back from the tracking server."""

import mlflow
from common import MLFLOW_TRACKING_URI, s3, spark
from pyspark.sql import functions as f

gold = spark().table("lakehouse.medallion.revenue_by_brand")

mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
mlflow.set_experiment("ghost-kitchen")
with mlflow.start_run(run_name="brand-revenue") as run:
    stats = gold.agg(
        f.count("*").alias("brands"),
        f.round(f.sum("revenue"), 2).alias("revenue"),
        f.max("avg_order").alias("best_avg_order"),
    ).first()
    mlflow.log_param("source_table", "lakehouse.medallion.revenue_by_brand")
    mlflow.log_metrics(stats.asDict())
    mlflow.log_text(gold.toPandas().to_csv(index=False), "revenue_by_brand.csv")

logged = mlflow.get_run(run.info.run_id)
print(f"{logged.info.run_name}: {logged.info.status}")
for name, value in sorted(logged.data.metrics.items()):
    print(f"  {name} = {value}")

# The artifact went through the tracking server into the bucket.
for obj in s3().list_objects_v2(Bucket="lakehouse", Prefix="mlflow/")["Contents"]:
    print(obj["Key"])

python log_run.py

2026/10/01 14:41:30 INFO mlflow.tracking.fluent: Experiment with name 'ghost-kitchen' does not exist. Creating a new experiment.
🏃 View run brand-revenue at: http://localhost:5000/#/experiments/1/runs/88457b52f19d43fea413f804bf7d6509
🧪 View experiment at: http://localhost:5000/#/experiments/1
brand-revenue: FINISHED
  best_avg_order = 74.92
  brands = 5.0
  revenue = 577384.27
mlflow/1/88457b52f19d43fea413f804bf7d6509/artifacts/revenue_by_brand.csv

services:
  seaweedfs:
    image: chrislusf/seaweedfs:3.80
    entrypoint: weed
    command: server -dir=/data -s3 -s3.config=/etc/seaweedfs/s3.json -volume.max=0 -master.volumeSizeLimitMB=1024
    ports:
      - "${S3_PORT:-8333}:8333"
    volumes:
      - ./s3.json:/etc/seaweedfs/s3.json:ro
      - seaweedfs-data:/data
    healthcheck:
      test: wget -qO /dev/null http://127.0.0.1:8333/status
      interval: 5s
      retries: 30

  spark-master:
    image: apache/spark:4.2.0
    hostname: spark-master
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.master.Master --host spark-master
    ports:
      - "${SPARK_UI_PORT:-8080}:8080"
    healthcheck:
      test: curl -sf http://localhost:8080 > /dev/null
      interval: 5s
      retries: 30

  spark-worker:
    image: apache/spark:4.2.0
    hostname: spark-worker
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.worker.Worker --host spark-worker spark://spark-master:7077
    environment:
      SPARK_WORKER_CORES: "6"
      SPARK_WORKER_MEMORY: 4g
    volumes:
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy

  spark-connect:
    image: apache/spark:4.2.0
    hostname: spark-connect
    command: >
      /opt/spark/sbin/start-connect-server.sh
      --master spark://spark-master:7077
      --packages io.delta:delta-spark_4.2_2.13:4.4.0,io.delta:delta-iceberg_2.13:4.4.0,io.unitycatalog:unitycatalog-spark_4.2_2.13:0.6.0,org.apache.hadoop:hadoop-aws:3.5.0,org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0
      --exclude-packages io.delta:delta-spark_4.1_2.13
    environment:
      SPARK_NO_DAEMONIZE: "1"
    ports:
      - "${SPARK_CONNECT_PORT:-15002}:15002"
    volumes:
      - ./spark-conf:/opt/spark/conf
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "bash", "-c", "echo > /dev/tcp/localhost/15002"]
      interval: 5s
      start_period: 15m

  unity-catalog:
    image: unitycatalog/unitycatalog:v0.6.0
    ports:
      - "${UC_PORT:-8081}:8080"
    volumes:
      - ./server.properties:/home/unitycatalog/etc/conf/server.properties:ro
      - uc-data:/home/unitycatalog/etc/db
    healthcheck:
      test: wget -qO /dev/null http://127.0.0.1:8080/api/2.1/unity-catalog/catalogs
      interval: 5s
      retries: 30

  kafka:
    image: apache/kafka:4.2.0
    ports:
      - "${KAFKA_PORT:-9092}:9094"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: INTERNAL://:9092,CONTROLLER://:9093,EXTERNAL://:9094
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://localhost:${KAFKA_PORT:-9092}
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
    healthcheck:
      test: /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
      interval: 10s
      retries: 30

  airflow:
    build:
      dockerfile_inline: |
        FROM apache/airflow:3.3.2-python3.12
        RUN pip install --no-cache-dir apache-airflow==3.3.2 apache-airflow-providers-apache-spark==6.3.2 pyspark-client==4.2.0
    command: standalone
    environment:
      AIRFLOW__CORE__LOAD_EXAMPLES: "false"
      AIRFLOW__CORE__SIMPLE_AUTH_MANAGER_ALL_ADMINS: "true"
      AIRFLOW_CONN_SPARK_CONNECT_DEFAULT: '{"conn_type": "spark_connect", "host": "spark-connect", "port": 15002}'
    ports:
      - "${AIRFLOW_PORT:-8085}:8080"
    volumes:
      - airflow-data:/opt/airflow
      - ./dags:/opt/airflow/dags
      - ./pipeline:/opt/airflow/pipeline
    healthcheck:
      test: curl -sf http://localhost:8080/api/v2/monitor/health
      interval: 5s
      start_period: 5m

  mlflow:
    build:
      dockerfile_inline: |
        FROM ghcr.io/mlflow/mlflow:v3.16.1
        RUN pip install --no-cache-dir boto3==1.43.93
    command: >
      mlflow server --host 0.0.0.0 --port 5000 --workers 1
      --backend-store-uri sqlite:////mlflow/mlflow.db
      --artifacts-destination s3://lakehouse/mlflow
    environment:
      MLFLOW_S3_ENDPOINT_URL: http://seaweedfs:8333
      AWS_ACCESS_KEY_ID: lakehouse
      AWS_SECRET_ACCESS_KEY: lakehouse-secret
    ports:
      - "${MLFLOW_PORT:-5000}:5000"
    volumes:
      - mlflow-data:/mlflow
    healthcheck:
      test: python -c "import urllib.request; urllib.request.urlopen('http://localhost:5000/health')"
      interval: 5s
      retries: 30

volumes:
  seaweedfs-data:
  spark-work:
  uc-data:
  airflow-data:
  mlflow-data:

docker compose ps --format "table {{.Service}}\t{{.Status}}"

SERVICE         STATUS
airflow         Up 37 seconds (healthy)
kafka           Up 4 minutes (healthy)
mlflow          Up 37 seconds (healthy)
seaweedfs       Up 6 minutes (healthy)
spark-connect   Up 5 minutes (healthy)
spark-master    Up 6 minutes (healthy)
spark-worker    Up 6 minutes
unity-catalog   Up 5 minutes (healthy)