06. 数据湖技术：Iceberg vs Delta Lake vs Hudi

-- Iceberg + Spark SQL
CREATE TABLE iceberg_db.orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(18,2),
    dt STRING
) USING iceberg
PARTITIONED BY (dt);

-- 原子性替换分区
CALL iceberg_catalog.system.replace_partition_field(
    'db.orders', 'dt', 'days(dt)', 'day'
);

-- MERGE INTO 原子更新
MERGE INTO iceberg_db.orders t
USING (SELECT * FROM staging_orders) s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET t.amount = s.amount
WHEN NOT MATCHED THEN INSERT *;

# Delta Lake + PySpark
from delta import configure_spark_with_delta_pip
from pyspark.sql import SparkSession

builder = SparkSession.builder \
    .appName("DeltaLake") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
spark = configure_spark_with_delta_pip(builder).getOrCreate()

# 创建表
df.write.format("delta").mode("overwrite").save("/delta/orders")

# MERGE (UPSERT)
from delta.tables import DeltaTable

deltaTable = DeltaTable.forPath(spark, "/delta/orders")
deltaTable.alias("t").merge(
    updates_df.alias("s"),
    "t.order_id = s.order_id"
).whenMatchedUpdate(set={"amount": "s.amount"}) \
 .whenNotMatchedInsert(values={"order_id": "s.order_id", "amount": "s.amount"}) \
 .execute()

# 乐观并发写入
df.write.format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .save("/delta/orders")

// Hudi + Spark
Map<String, String> options = new HashMap<>();
options.put("hoodie.table.name", "orders");
options.put("hoodie.datasource.write.recordkey.field", "order_id");
options.put("hoodie.datasource.write.partitionpath.field", "dt");
options.put("hoodie.datasource.write.table.type", "MERGE_ON_READ"); // MOR
options.put("hoodie.datasource.write.operation", "upsert");
options.put("hoodie.datasource.write.precombine.field", "ts");

df.write()
    .format("hudi")
    .options(options)
    .mode(SaveMode.Append)
    .save("s3://bucket/hudi/orders");

-- 1. 添加列
ALTER TABLE iceberg_db.orders ADD COLUMN shipping_address STRING;

-- 2. 安全类型提升（INT → BIGINT）
ALTER TABLE iceberg_db.orders ALTER COLUMN amount TYPE BIGINT;

-- 3. 重命名列
ALTER TABLE iceberg_db.orders RENAME COLUMN amount TO total_amount;

-- 4. 嵌套结构变更（STRUCT / MAP / LIST 内部）
ALTER TABLE iceberg_db.orders 
ADD COLUMN products AFTER shipping_address;

-- 5. 分区演化（无需重写历史数据！）
ALTER TABLE iceberg_db.orders 
ADD PARTITION FIELD bucket(16, user_id);

-- 查询旧快照 → 自动使用旧 Schema
SELECT * FROM iceberg_db.orders TIMESTAMP AS OF '2024-01-01 00:00:00';

# 自动 Schema 演化
df.write.format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .save("/delta/orders")

# 显式添加列
from pyspark.sql.functions import lit
spark.read.format("delta").load("/delta/orders") \
    .withColumn("new_field", lit(None)) \
    .write.format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .save("/delta/orders")

-- 查询历史快照
SELECT * FROM iceberg_db.orders FOR SYSTEM_VERSION AS OF 123456789;
SELECT * FROM iceberg_db.orders FOR SYSTEM_TIME AS OF '2024-06-01 00:00:00';

-- 查看所有快照
SELECT * FROM iceberg_db.orders.snapshots;

-- 回滚到指定快照
CALL iceberg_catalog.system.rollback_to_snapshot('db.orders', 123456789);

-- 设置快照过期（保留策略）
ALTER TABLE iceberg_db.orders SET TBLPROPERTIES (
    'history.expire.max-snapshot-age-ms' = '604800000',  -- 7天
    'history.expire.min-snapshots-to-keep' = '5'
);

# 按版本号读
df_v0 = spark.read.format("delta").option("versionAsOf", 0).load("/delta/orders")
df_v5 = spark.read.format("delta").option("versionAsOf", 5).load("/delta/orders")

# 按时间戳读
df_ts = spark.read.format("delta").option("timestampAsOf", "2024-06-01T00:00:00Z").load("/delta/orders")

# 查看历史版本
spark.sql("DESCRIBE HISTORY delta.`/delta/orders`").show()

# 回滚
deltaTable = DeltaTable.forPath(spark, "/delta/orders")
deltaTable.restoreToVersion(0)  # 或 restoreToTimestamp

# Vacuum 清理过期文件（保留 7 天）
spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")
spark.sql("VACUUM delta.`/delta/orders` RETAIN 168 HOURS")

// Hudi 增量/时间旅行查询
Dataset<Row> incrementalDF = spark.read()
    .format("hudi")
    .option("hoodie.datasource.query.type", "incremental")
    .option("hoodie.datasource.read.begin.instanttime", "20240101000000")
    .option("hoodie.datasource.read.end.instanttime", "20240102000000")
    .load("s3://bucket/hudi/orders");

// MOR 表读优化（读取已合并的数据）
Dataset<Row> readOptimized = spark.read()
    .format("hudi")
    .option("hoodie.datasource.query.type", "read_optimized")
    .load("s3://bucket/hudi/orders");

# spark-defaults.conf
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.iceberg_catalog=org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.iceberg_catalog.type=hive
spark.sql.catalog.iceberg_catalog.uri=thrift://hive-metastore:9083
spark.sql.catalog.iceberg_catalog.warehouse=s3://bucket/iceberg-warehouse

# 表级优化配置
spark.sql("""
    CREATE TABLE iceberg_catalog.db.orders (
        order_id BIGINT,
        user_id BIGINT,
        amount DECIMAL(18,2),
        ts TIMESTAMP,
        dt DATE
    ) USING iceberg
    PARTITIONED BY (days(ts))
    TBLPROPERTIES (
        'write_compression' = 'ZSTD',
        'write_metadata_compression' = 'GZIP',
        'commit.manifest.min-count-to-merge' = '5',
        'history.expire.max-snapshot-age-ms' = '604800000'
    )
""")

# 自动优化与压缩
spark.sql("""
    CREATE TABLE delta.`/delta/orders` (
        order_id BIGINT,
        user_id BIGINT,
        amount DECIMAL(18,2)
    ) USING DELTA
    TBLPROPERTIES (
        'delta.autoOptimize.optimizeWrite' = 'true',
        'delta.autoOptimize.autoCompact' = 'true',
        'delta.deletedFileRetentionDuration' = 'interval 7 days',
        'delta.logRetentionDuration' = 'interval 30 days'
    )
""")

# CDF (Change Data Feed) 开启变更追踪
spark.sql("""
    ALTER TABLE delta.`/delta/orders` 
    SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

# 读取 CDC
cdc_df = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 0) \
    .load("/delta/orders")

# 对象存储目录布局（S3 / OSS / GCS）
warehouse/
  ├── bronze/
  │   ├── raw_orders/           # Iceberg 格式，近实时摄入
  │   ├── raw_events/
  │   └── raw_logs/             # 原始日志保留 30 天
  ├── silver/
  │   ├── cleaned_orders/       # 去重、标准化后的订单
  │   ├── user_sessions/        # 会话聚合
  │   └── product_inventory/    # 库存快照（含 Schema Evolution）
  └── gold/
      ├── daily_revenue/        # 日度收入报表表
      ├── user_ltv/             # 用户生命周期价值
      └── ml_feature_store/     # 特征工程输出

-- 按版本号查询历史数据
SELECT * FROM delta.`/delta/orders` VERSION AS OF 5;

-- 按时间戳查询
SELECT * FROM delta.`/delta/orders` TIMESTAMP AS OF '2024-06-15T00:00:00Z';

-- 查看完整历史
DESCRIBE HISTORY delta.`/delta/orders`;

-- 回滚到指定版本（生成新的反向提交，不删除历史）
RESTORE TABLE delta.`/delta/orders` TO VERSION AS OF 3;

# 流式写入 Delta（exactly-once）
stream_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/delta/checkpoints/orders") \
    .start("/delta/orders")

# 同一表支持批式回溯写入
batch_df.write.format("delta") \
    .mode("overwrite") \
    .option("replaceWhere", "dt >= '2024-06-01' and dt < '2024-07-01'") \
    .save("/delta/orders")

-- 对 user_id 与 product_id 执行 Z-Order 优化
OPTIMIZE delta.`/delta/orders`
ZORDER BY (user_id, product_id);

-- 查看优化效果（文件跳过统计）
DESCRIBE DETAIL delta.`/delta/orders`;

-- 创建按月隐藏分区的表
CREATE TABLE iceberg_catalog.db.events (
    event_id BIGINT,
    event_time TIMESTAMP,
    user_id STRING
) USING iceberg
PARTITIONED BY (months(event_time));

-- 查询时无需带分区 filter，优化器自动下推
SELECT * FROM iceberg_catalog.db.events
WHERE event_time >= '2024-01-01' AND event_time < '2024-02-01';

-- 分区演进：无需重写历史数据即可添加新分区策略
ALTER TABLE iceberg_catalog.db.events
ADD PARTITION FIELD bucket(16, user_id);

-- 行级删除（生成 position-delete 文件，不重写 Parquet）
DELETE FROM iceberg_catalog.db.orders WHERE status = 'cancelled';

-- MERGE INTO 实现 Upsert
MERGE INTO iceberg_catalog.db.orders t
USING staging_orders 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
WHEN NOT MATCHED THEN INSERT *;

# PySpark 读取 Hudi 增量数据
incremental_df = spark.read \
    .format("hudi") \
    .option("hoodie.datasource.query.type", "incremental") \
    .option("hoodie.datasource.read.begin.instanttime", "20240901000000") \
    .option("hoodie.datasource.read.end.instanttime", "20240902000000") \
    .load("s3://bucket/hudi/orders")

# 读取实时视图（含未 compaction 的增量 log）
realtime_df = spark.read \
    .format("hudi") \
    .option("hoodie.datasource.query.type", "snapshot") \
    .load("s3://bucket/hudi/orders")

# 读取读优化视图（仅已 compaction 的 Base 文件）
ro_df = spark.read \
    .format("hudi") \
    .option("hoodie.datasource.query.type", "read_optimized") \
    .load("s3://bucket/hudi/orders")

# hudi-deltastreamer.properties
hoodie.deltastreamer.source.kafka.topic=db.orders.cdc
hoodie.deltastreamer.schemaprovider.registry.url=http://schema-registry:8081/subjects/db.orders-value/versions/latest
hoodie.datasource.write.recordkey.field=order_id
hoodie.datasource.write.partitionpath.field=dt
hoodie.datasource.write.table.type=MERGE_ON_READ
hoodie.datasource.write.precombine.field=update_ts
hoodie.compact.inline=false
hoodie.compact.schedule.inline=true

# DataHub 元数据摄取示例（Iceberg 表血缘）
from datahub.ingestion.api.source import Source
from datahub.ingestion.run.pipeline import Pipeline

# 配置 Iceberg REST Catalog 连接器
config = {
    "source": {
        "type": "iceberg",
        "config": {
            "catalog": {
                "type": "rest",
                "uri": "http://iceberg-rest:8181"
            },
            "profiling": {
                "enabled": True,
                "include_column_stats": True
            }
        }
    },
    "sink": {
        "type": "datahub-rest",
        "config": {
            "server": "http://datahub-gms:8080"
        }
    }
}

pipeline = Pipeline.create(config)
pipeline.run()
pipeline.raise_from_status()