湖仓一体架构：Iceberg、Delta Lake 与 Hudi 的统一数据底座

# iceberg_catalog.py
from pyiceberg.catalog import load_catalog

catalog = load_catalog(
    "rest",
    **{
        "uri": "http://iceberg-rest:8181",
        "warehouse": "s3://data-lake/warehouse",
        "s3.endpoint": "http://minio:9000",
    },
)

# 列出命名空间并创建订单域命名空间
print(catalog.list_namespaces())
catalog.create_namespace("ods")

table = catalog.create_table(
    "ods.orders",
    schema={
        "order_id": "string",
        "user_id": "long",
        "amount": "double",
        "status": "string",
        "event_time": "timestamp",
    },
    partition_spec="day(event_time)",
    properties={"write.format.default": "parquet"},
)

-- 基于快照 ID 读取历史版本
SELECT * FROM ods.orders
  VERSION AS OF 8201468987888888888
WHERE event_time >= '2026-09-01';

-- 基于时间戳读取
SELECT * FROM ods.orders
  TIMESTAMP AS OF '2026-09-25 08:00:00'
LIMIT 100;

-- 合并小文件并按 dt, user_id 聚类
OPTIMIZE dws.daily_order_revenue
ZORDER BY (dt, user_id);

-- 清理超过保留期的历史版本（默认 7 天）
VACUUM dws.daily_order_revenue RETAIN 168 HOURS;

-- 创建 MOR 表并做流式 Upsert
CREATE TABLE ods.orders_hudi (
  order_id STRING,
  status STRING,
  amount DOUBLE,
  ts TIMESTAMP
) USING hudi
TBLPROPERTIES (
  'hoodie.table.type' = 'MERGE_ON_READ',
  'hoodie.datasource.write.recordkey.field' = 'order_id',
  'hoodie.datasource.write.precombine.field' = 'ts'
);

-- 增量视图读取最近一次提交的变更
SELECT * FROM hudi_incremental('ods.orders_hudi', 'beginTime', '20260925120000');

# medallion_pipeline.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_timestamp

spark = (
    SparkSession.builder.appName("medallion_lakehouse")
    .config("spark.sql.catalog.lake", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.lake.type", "rest")
    .config("spark.sql.catalog.lake.uri", "http://iceberg-rest:8181")
    .getOrCreate()
)

stream = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "orders.raw")
    .load()
)

# Bronze：原样落湖
bronze = (
    stream.selectExpr("CAST(value AS STRING) AS raw", "to_timestamp(timestamp) AS event_time")
    .writeStream.format("iceberg")
    .option("path", "lake.ods.orders_bronze")
    .trigger(processingTime="60 seconds")
    .checkpointLocation("s3://data-lake/checkpoints/orders_bronze")
)

# Silver：清洗后写入（并行流）
silver = (
    stream.selectExpr("CAST(value AS STRING) AS raw")
    .select(parse_json(col("raw")))
    .filter(col("status").isNotNull())
    .writeStream.format("iceberg")
    .option("path", "lake.ods.orders_silver")
    .trigger(processingTime="60 seconds")
    .checkpointLocation("s3://data-lake/checkpoints/orders_silver")
)

-- Trino: 读取 S3 上的 Iceberg 表
CALL iceberg.system.register_table(
    schema_name => 'ods',
    table_name => 'orders',
    table_location => 's3://data-lake/warehouse/ods/orders'
);

SELECT dt, count(*) AS orders
FROM "lake.ods".orders
WHERE dt >= '2026-09-20'
GROUP BY dt
ORDER BY dt;

# flink_iceberg.yaml
job:
  name: orders_cdc_to_iceberg
  checkpoint:
    interval: 60s
    mode: exactly_once
  source:
    connector: debezium
    database: mysql
    tables: [shop.orders]
  sink:
    connector: iceberg
    catalog-name: lake
    table: lake.ods.orders
    write:
      format: parquet
      distribution-mode: hash
      upsert-enabled: true

-- 合并 orders_silver 中小于 32MB 的数据文件
CALL lake.system.rewrite_data_files(
    table => 'ods.orders_silver',
    strategy => 'binpack',
    min_file_size_in_bytes => 33554432,
    max_file_size_in_bytes => 134217728
);

-- 保留 7 天内快照，清理更早历史
CALL lake.system.expire_snapshots(
    table => 'ods.orders_silver',
    older_than => TIMESTAMP '2026-09-18 00:00:00',
    retain_last => 10
);

-- 清理无引用的孤儿文件
CALL lake.system.remove_orphan_files(
    table => 'ods.orders_silver',
    older_than => TIMESTAMP '2026-09-24 00:00:00'
);

# zorder_write.py
from pyspark.sql.functions import col

# 用 sortWithinPartitions 实现轻量 Z-Order 效果
(
    spark.table("lake.ods.orders_silver")
    .repartitionByRange(col("dt"))
    .sortWithinPartitions(col("user_id"), col("order_id"))
    .writeTo("lake.ods.orders_silver_z")
    .using("iceberg")
    .overwritePartitions()
)

MERGE INTO lake.ods.orders_silver t
USING (
  SELECT order_id, user_id, amount, status, op, event_time
  FROM lake.ods.orders_cdc
) s
ON t.order_id = s.order_id
WHEN MATCHED AND s.op = 'D' THEN DELETE
WHEN MATCHED THEN UPDATE SET
  t.amount = s.amount, t.status = s.status, t.event_time = s.event_time
WHEN NOT MATCHED AND s.op = 'I' THEN
  INSERT (order_id, user_id, amount, status, event_time)
  VALUES (s.order_id, s.user_id, s.amount, s.status, s.event_time);

#!/bin/bash
# reconcile.sh
# 对比流式 Silver 表与日批 Gold 表的订单数
DIFF=$(trino --execute \
  "SELECT
     (SELECT count(*) FROM \"lake.ods\".orders_silver WHERE dt = '2026-09-25') -
     (SELECT count(*) FROM \"lake.dws\".daily_order_revenue WHERE dt = '2026-09-25')")

if [ "$DIFF" != "0" ]; then
  echo "RECONCILE FAILED: diff=$DIFF"
  exit 1
fi

-- Hive 表原地转 Iceberg
CALL lake.system.migrate('ods.orders_legacy');

-- Delta 表转 Iceberg（Spark 3.x）
ALTER TABLE lake.ods.orders
  SET TBLPROPERTIES ('engine' = 'iceberg');