DuckLake: สร้าง Lakehouse บน DuckDB ง่ายๆ ในเวลาไม่ถึงชั่วโมง

pip install "duckdb>=1.5.2"
python -c "import duckdb; print(duckdb.__version__)"

# step1_make_data.py
import os
 
import duckdb
 
os.makedirs("raw", exist_ok=True)
con = duckdb.connect()
 
# Two months of fake orders, 5,000 rows each, written as plain Parquet files.
for month, first_day, first_id in [("2026_01", "2026-01-01", 0), ("2026_02", "2026-02-01", 5000)]:
    con.execute(f"""
        COPY (
            SELECT
                {first_id} + i AS order_id,
                DATE '{first_day}' + CAST(i % 28 AS INTEGER) AS order_date,
                ['north', 'south', 'east', 'west'][1 + i % 4] AS region,
                ['pending', 'shipped', 'delivered'][1 + i % 3] AS status,
                CAST(10 + (i * 37 % 900) / 10.0 AS DECIMAL(10, 2)) AS amount
            FROM range(5000) AS t(i)
        ) TO 'raw/orders_{month}.parquet' (FORMAT parquet)
    """)
 
con.sql("SELECT count(*) AS rows, min(order_date), max(order_date) FROM 'raw/orders_*.parquet'").show()

# common.py
import json
import os
 
import duckdb
 
MARKERS = "lake/markers.json"
 
def connect():
    """Open a DuckDB session with the DuckLake attached and selected."""
    os.makedirs("lake/data", exist_ok=True)       # the folder for the catalog file must already exist
    con = duckdb.connect()
    con.execute("INSTALL ducklake")
    con.execute("LOAD ducklake")
    if os.path.exists("lake/metadata.ducklake"):
        con.execute("ATTACH 'ducklake:lake/metadata.ducklake' AS lake")   # data path is stored in the catalog
    else:
        con.execute("ATTACH 'ducklake:lake/metadata.ducklake' AS lake (DATA_PATH 'lake/data/')")
    con.execute("USE lake")
    return con
 
def mark(con, name):
    """Remember the newest snapshot ID under a name, so later steps can travel back to it."""
    snapshot_id = con.execute("SELECT max(snapshot_id) FROM lake.snapshots()").fetchone()[0]
    markers = markers_so_far()
    markers[name] = snapshot_id
    with open(MARKERS, "w") as f:
        json.dump(markers, f)
    return snapshot_id
 
def markers_so_far():
    if os.path.exists(MARKERS):
        with open(MARKERS) as f:
            return json.load(f)
    return {}

# step2_load.py
from common import connect, mark
 
con = connect()
 
# Create the table straight from the January Parquet file.
con.execute("CREATE TABLE orders AS SELECT * FROM read_parquet('raw/orders_2026_01.parquet')")
mark(con, "loaded")
 
con.sql("SELECT count(*) AS rows, min(order_date) AS first_day, max(order_date) AS last_day FROM orders").show()
con.sql("FROM lake.snapshots()").show()
con.sql("FROM glob('lake/data/**/*')").show()

# step3_change.py
from common import connect, mark
 
con = connect()
 
# 1. Append February.
con.execute("INSERT INTO orders SELECT * FROM read_parquet('raw/orders_2026_02.parquet')")
mark(con, "appended")
 
# 2. Update many rows: early-January shipped orders are now delivered.
con.execute("UPDATE orders SET status = 'delivered' WHERE status = 'shipped' AND order_date < DATE '2026-01-15'")
mark(con, "updated")
 
# 3. Change the schema: new column, renamed column.
con.execute("ALTER TABLE orders ADD COLUMN channel VARCHAR")
con.execute("ALTER TABLE orders RENAME COLUMN amount TO total")
mark(con, "evolved")
 
con.sql("SELECT snapshot_id, snapshot_time, changes FROM lake.snapshots() ORDER BY snapshot_id").show()
con.sql("SELECT * FROM orders LIMIT 3").show()

# step4_time_travel.py
from common import connect, mark, markers_so_far
 
con = connect()
m = markers_so_far()
 
# Row counts at three points in history.
for name in ["loaded", "appended", "evolved"]:
    n = con.execute(f"SELECT count(*) FROM orders AT (VERSION => {m[name]})").fetchone()[0]
    print(f"{name:<9} (snapshot {m[name]}): {n} rows")
 
# The schema is versioned too: the old snapshot still has the column called amount.
con.sql(f"SELECT * FROM orders AT (VERSION => {m['loaded']}) LIMIT 3").show()
 
# Make a mistake, then undo it from history.
con.execute("DELETE FROM orders WHERE region = 'west'")
mark(con, "mistake")
 
con.execute(f"""
    INSERT INTO orders (order_id, order_date, region, status, total)
    SELECT order_id, order_date, region, status, total
    FROM orders AT (VERSION => {m['evolved']})
    WHERE region = 'west'
""")

# step5_maintenance.py
from common import connect
 
con = connect()
 
# Every write made new files. Merge small adjacent files into larger ones.
con.sql("SELECT * FROM ducklake_merge_adjacent_files('lake')").show()
con.sql("FROM glob('lake/data/**/*')").show()

ATTACH 'ducklake:postgres:dbname=ducklake_catalog host=localhost' AS lake (DATA_PATH 'lake/data/');