数据工程的专业底线：分层建模、增量合并、质量断言与调度闭环，附可运行代码

-- DWD层：订单明细事实表（只存外键+度量）
CREATE TABLE dwd_order_detail (
  order_id     STRING,
  user_sk      BIGINT,   -- 维度代理键，不是业务ID
  product_sk   BIGINT,
  order_amt    DECIMAL(18,2),
  qty          INT,
  dt           STRING
) PARTITIONED BY (dt);

-- 维度表：用户维（SCD2缓慢变化维）
CREATE TABLE dim_user (
  user_sk      BIGINT,   -- 代理键
  user_id      STRING,   -- 业务键
  city         STRING,
  level        STRING,
  start_dt     STRING,
  end_dt       STRING,
  is_current   BOOLEAN
);

# scd2_merge.py  —— PySpark实现SCD2
from pyspark.sql import functions as F

def scd2_merge(spark, dim_table, inc_df):
    cur = spark.table(dim_table)
    # 1. 关闭已变化的当前版本
    changed = inc_df.alias("i").join(
        cur.filter("is_current=true").alias("c"), "user_id"
    ).where("i.level <> c.level or i.city <> c.city") \
     .select("c.user_sk", "c.user_id")

    closed = cur.join(changed, "user_sk", "left_anti") \
        .union(changed.join(cur, "user_sk")
               .withColumn("end_dt", F.current_date().cast("string"))
               .withColumn("is_current", F.lit(False)))

    # 2. 插入新版本（代理键用row_number+偏移生成，生产用sequence）
    new_ver = inc_df.join(cur.filter("is_current=true").select("user_id"), "user_id", "left_anti") \
        .withColumn("user_sk", F.monotonically_increasing_id()) \
        .withColumn("start_dt", F.current_date().cast("string")) \
        .withColumn("end_dt", F.lit("9999-12-31")) \
        .withColumn("is_current", F.lit(True))

    closed.union(new_ver).write.mode("overwrite").saveAsTable(dim_table)

MERGE INTO dwd_order_detail t
USING (SELECT * FROM stg_order WHERE dt = '${biz_date}') s
ON t.order_id = s.order_id AND t.dt = s.dt
WHEN MATCHED AND t.order_amt <> s.order_amt THEN UPDATE SET
  order_amt = s.order_amt, qty = s.qty
WHEN NOT MATCHED THEN INSERT (order_id,user_sk,product_sk,order_amt,qty,dt)
VALUES (s.order_id,s.user_sk,s.product_sk,s.order_amt,s.qty,s.dt);

# quality.py
from great_expectations.dataset import PandasDataset

def validate(df, biz_date):
    ds = PandasDataset(df)
    ds.expect_column_values_to_not_be_null("order_id")
    ds.expect_column_values_to_be_unique("order_id")
    ds.expect_column_values_to_be_between("order_amt", 0, 1_000_000)
    # 主键唯一率、空值率、金额分布偏离
    res = ds.validate()
    if not res["success"]:
        raise ValueError(f"[{biz_date}] 质量门禁失败: {res['statistics']}")
    return True

# dag_order_dwd.py —— Airflow DAG
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default = {"retries": 3, "retry_delay": timedelta(minutes=5),
           "depends_on_past": False, "max_active_runs": 1}

with DAG("order_dwd", start_date=datetime(2024,1,1),
         schedule="0 2 * * *", default_args=default,
         catchup=True) as dag:
    extract = PythonOperator(task_id="ods_extract", python_callable=run_ods,
                             op_kwargs={"ds": "{{ ds }}"})
    merge   = PythonOperator(task_id="dwd_merge", python_callable=run_dwd,
                             op_kwargs={"ds": "{{ ds }}"})
    check   = PythonOperator(task_id="quality_gate", python_callable=run_check,
                             op_kwargs={"ds": "{{ ds }}"})
    extract >> merge >> check

def log_lineage(task_id, inputs, outputs, ds):
    db.lineage.insert_one({"task": task_id, "inputs": inputs,
                           "outputs": outputs, "ds": ds,
                           "run_at": datetime.utcnow()})