云原生数据仓库AnalyticDB：基于AnalyticDB Spark+Notebook实现数仓与数据湖间的数据交换

Spark SQL

执行前需创建一个绑定 OSS 路径的数据库，作为数据湖的存储根目录，可以让后续的表自动继承路径。后续建表时无需再次指定路径，可以更好地区分湖仓表（额外声明创建一个库用于管理湖表）。

CREATE DATABASE IF NOT EXISTS LAKEHOUSE_DB LOCATION 'oss://<Bucket>/ORDERS_DATALAKE_SNAPSHOT/';

CREATE TABLE IF NOT EXISTS LAKEHOUSE_DB.ORDERS_DATALAKE_SNAPSHOT 
USING ICEBERG
AS SELECT * FROM jdbc.DATAWAREHOUSE.ORDERS;

CREATE TABLE IF NOT EXISTS LAKEHOUSE_DB.LINEITEM_DATALAKE_SNAPSHOT 
USING ICEBERG
AS SELECT * FROM jdbc.DATAWAREHOUSE.LINEITEM LIMIT 10000;

XIHE SQL

替换LOCATION为OSS实际Bucket。

-- 快照 Orders 表
CREATE TABLE IF NOT EXISTS DATAWAREHOUSE.ORDERS_DATALAKE_SNAPSHOT 
STORED AS ICEBERG
LOCATION 'oss://<Bucket>/ORDERS_DATALAKE_SNAPSHOT/'
AS SELECT * FROM DATAWAREHOUSE.ORDERS;

-- 快照 Lineitem 表
CREATE TABLE IF NOT EXISTS DATAWAREHOUSE.LINEITEM_DATALAKE_SNAPSHOT 
STORED AS ICEBERG
LOCATION 'oss://<Bucket>/ORDERS_DATALAKE_SNAPSHOT/'
AS SELECT * FROM DATAWAREHOUSE.LINEITEM;

Spark SQL

INSERT OVERWRITE LAKEHOUSE_DB.order_fulfillment_detail
SELECT 
    CONCAT(CAST(o.o_orderkey AS STRING), '-', CAST(l.l_linenumber AS STRING)) AS transaction_id,
    o.o_orderkey            AS order_key,
    l.l_linenumber          AS line_number,
    o.o_orderdate           AS order_date,
    l.l_shipdate            AS ship_date,
    l.l_commitdate          AS commit_date,
    DATEDIFF(l.l_shipdate, o.o_orderdate) AS fulfillment_days,
    CASE 
        WHEN l.l_shipdate > l.l_commitdate THEN true 
        ELSE false 
    END                     AS is_delayed,
    l.l_shipmode            AS ship_mode,
    o.o_custkey             AS customer_key,
    o.o_clerk               AS clerk_id,
    o.o_orderpriority       AS priority,
    o.o_orderstatus         AS order_status,
    l.l_linestatus          AS line_status,
    l.l_returnflag          AS return_flag,
    l.l_quantity            AS quantity,
    l.l_extendedprice       AS list_amount,
    CAST(l.l_extendedprice * l.l_discount AS DECIMAL(18, 2)) AS discount_amount,
    CAST(l.l_extendedprice * (1 - l.l_discount) * l.l_tax AS DECIMAL(18, 2)) AS tax_amount,
    CAST(l.l_extendedprice * (1 - l.l_discount) * (1 + l.l_tax) AS DECIMAL(18, 2)) AS final_pay_amount
FROM LAKEHOUSE_DB.ORDERS_DATALAKE_SNAPSHOT o
JOIN LAKEHOUSE_DB.LINEITEM_DATALAKE_SNAPSHOT l 
    ON o.o_orderkey = l.l_orderkey
WHERE 
    -- 时间过滤，减少 80% 以上的数据扫描
    o.o_orderdate >= '1998-01-01' 
    -- 状态过滤，只看已下单的活跃订单
    AND o.o_orderstatus IN ('O', 'F') 
    -- 进一步缩小范围（只分析高优先级的订单）
    AND o.o_orderpriority IN ('1-URGENT', '2-HIGH');

PySpark

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, lit, concat, concat_ws, datediff, 
    when, round, count, sum, avg
)
from pyspark.sql.types import DecimalType


# ==========================================
# 1. 从数据仓库表的Iceberg快照表里，读取源数据 (Reader)
# ==========================================
df_orders = spark.table("LAKEHOUSE_DB.ORDERS_DATALAKE_SNAPSHOT").alias("o")
df_lineitem = spark.table("LAKEHOUSE_DB.LINEITEM_DATALAKE_SNAPSHOT").alias("l")

# ==========================================
# 2. 核心逻辑 (Transformation)
# ==========================================

# 关联逻辑
df_joined = df_orders.join(
    df_lineitem, 
    on=col("o.o_orderkey") == col("l.l_orderkey"), 
    how="inner")

# 计算与字段映射逻辑
df_result = df_joined.select(
    # --- 唯一标识 ---
    # 使用 concat_ws (With Separator) 拼接主键，比 SQL 的 concat 更优雅
    concat_ws("-", col("o.o_orderkey"), col("l.l_linenumber")).alias("transaction_id"),
    col("o.o_orderkey").alias("order_key"),
    col("l.l_linenumber").alias("line_number"),

    # --- 时间与履约 ---
    col("o.o_orderdate").alias("order_date"),
    col("l.l_shipdate").alias("ship_date"),
    col("l.l_commitdate").alias("commit_date"),
    
    # 履约耗时 (Datediff)
    datediff(col("l.l_shipdate"), col("o.o_orderdate")).alias("fulfillment_days"),
    
    # 是否逾期 (Boolean 逻辑)
    (col("l.l_shipdate") > col("l.l_commitdate")).alias("is_delayed"),
    
    col("l.l_shipmode").alias("ship_mode"),

    # --- 业务属性 ---
    col("o.o_custkey").alias("customer_key"),
    col("o.o_clerk").alias("clerk_id"),
    col("o.o_orderpriority").alias("priority"),
    col("o.o_orderstatus").alias("order_status"),
    col("l.l_linestatus").alias("line_status"),
    col("l.l_returnflag").alias("return_flag"),

    # --- 财务指标 (注意类型转换) ---
    col("l.l_quantity").cast(DecimalType(10, 2)).alias("quantity"),
    col("l.l_extendedprice").cast(DecimalType(18, 2)).alias("list_amount"),

    # 折扣金额 = Price * Discount
    (col("l.l_extendedprice") * col("l.l_discount"))
        .cast(DecimalType(18, 2)).alias("discount_amount"),

    # 税额 = Price * (1 - Discount) * Tax
    (col("l.l_extendedprice") * (1 - col("l.l_discount")) * col("l.l_tax"))
        .cast(DecimalType(18, 2)).alias("tax_amount"),

    # 实付金额 = Price * (1 - Discount) * (1 + Tax)
    (col("l.l_extendedprice") * (1 - col("l.l_discount")) * (1 + col("l.l_tax")))
        .cast(DecimalType(18, 2)).alias("final_pay_amount")
)

# ==========================================
# 3. 写入数据到数据湖 (Writer)
# ==========================================
TARGET_TABLE = "LAKEHOUSE_DB.order_fulfillment_detail"

print(f"正在写入表: {TARGET_TABLE} ...")

df_result.writeTo(TARGET_TABLE) \
    .overwritePartitions() # 或者 .overwrite() 视需求而定

print(">>> ETL 完成！")

Spark SQL

INSERT INTO jdbc.DATAWAREHOUSE.product_inventory_analysis
SELECT 
    -- 1. 生成SKU主键
    CONCAT(CAST(p.p_partkey AS STRING), '-', CAST(ps.ps_suppkey AS STRING)) AS sku_id,
    p.p_partkey             AS part_key,
    ps.ps_suppkey           AS supplier_key,
    -- 2. 商品维度
    p.p_name                AS part_name,
    p.p_brand               AS brand,
    p.p_type                AS type,
    p.p_size                AS size,
    p.p_container           AS container,
    -- 3. 基础度量
    ps.ps_availqty          AS stock_qty,
    ps.ps_supplycost        AS supply_cost,
    p.p_retailprice         AS retail_price,
    -- 4. 资金占用 (Inventory Value)
    CAST(ps.ps_availqty * ps.ps_supplycost AS DECIMAL(18, 2)) AS total_inv_cost,
    -- 5. 潜在营收 (Potential GMV)
    CAST(ps.ps_availqty * p.p_retailprice AS DECIMAL(18, 2))  AS potential_revenue,
    -- 6. 单件毛利 (Unit Margin)
    CAST(p.p_retailprice - ps.ps_supplycost AS DECIMAL(18, 2)) AS unit_margin_amt,
    -- 7. 毛利率 (Margin Rate)
    CAST(
    CASE 
        WHEN p.p_retailprice = 0 THEN 0 
        ELSE (p.p_retailprice - ps.ps_supplycost) / p.p_retailprice 
    END 
    AS DECIMAL(10, 4))      AS margin_rate,
    -- 8. 倒挂风险判断
    CASE 
        WHEN ps.ps_supplycost > p.p_retailprice THEN true 
        ELSE false 
    END                     AS is_underwater
FROM LAKEHOUSE.part p
JOIN LAKEHOUSE.partsupp ps 
ON p.p_partkey = ps.ps_partkey;

PySpark

from pyspark.sql import SparkSession
from pyspark.sql.functions import (col, lit, concat, when, cast)
from pyspark.sql.types import DecimalType, BooleanType

df_part = spark.table("LAKEHOUSE.part").alias("p")
df_partsupp = spark.table("LAKEHOUSE.partsupp").alias("ps")


df_joined = df_part.join(
    df_partsupp,
    on=col("p.p_partkey") == col("ps.ps_partkey"),
    how="inner"
)

# 核心计算与转换逻辑
df_result = df_joined.select(
    # --- 1. 生成 SKU 主键 ---
    # SQL: CONCAT(CAST(p.p_partkey AS STRING), '-', CAST(ps.ps_suppkey AS STRING))
    concat(
        col("p.p_partkey").cast("string"), 
        lit("-"), 
        col("ps.ps_suppkey").cast("string")
    ).alias("sku_id"),
    
    col("p.p_partkey").alias("part_key"),
    col("ps.ps_suppkey").alias("supplier_key"),

    # --- 2. 商品维度 ---
    col("p.p_name").alias("part_name"),
    col("p.p_brand").alias("brand"),
    col("p.p_type").alias("type"),
    col("p.p_size").alias("size"),
    col("p.p_container").alias("container"),

    # --- 3. 基础度量 ---
    col("ps.ps_availqty").alias("stock_qty"),
    col("ps.ps_supplycost").alias("supply_cost"),
    col("p.p_retailprice").alias("retail_price"),

    # --- 4. 资金占用 (Inventory Value) ---
    # SQL: CAST(ps.ps_availqty * ps.ps_supplycost AS DECIMAL(18, 2))
    (col("ps.ps_availqty") * col("ps.ps_supplycost"))
        .cast(DecimalType(18, 2)).alias("total_inv_cost"),

    # --- 5. 潜在营收 (Potential GMV) ---
    # SQL: CAST(ps.ps_availqty * p.p_retailprice AS DECIMAL(18, 2))
    (col("ps.ps_availqty") * col("p.p_retailprice"))
        .cast(DecimalType(18, 2)).alias("potential_revenue"),

    # --- 6. 单件毛利 (Unit Margin) ---
    # SQL: CAST(p.p_retailprice - ps.ps_supplycost AS DECIMAL(18, 2))
    (col("p.p_retailprice") - col("ps.ps_supplycost"))
        .cast(DecimalType(18, 2)).alias("unit_margin_amt"),

    # --- 7. 毛利率 (Margin Rate) ---
    # 必须处理分母为 0 的情况
    # SQL: CASE WHEN p.p_retailprice = 0 THEN 0 ELSE (P-C)/P END
    when(col("p.p_retailprice") == 0, lit(0))
        .otherwise(
            (col("p.p_retailprice") - col("ps.ps_supplycost")) / col("p.p_retailprice")
        ).cast(DecimalType(10, 4)).alias("margin_rate"),

    # --- 8. 倒挂风险判断 ---
    # SQL: CASE WHEN cost > price THEN true ELSE false END
    when(col("ps.ps_supplycost") > col("p.p_retailprice"), lit(True))
        .otherwise(lit(False))
        .alias("is_underwater")
)

# ==========================================
# 写入数据仓库表 (DATAWAREHOUSE 层)
# ==========================================
# 根据 SQL 的 INSERT INTO jdbc.DATAWAREHOUSE... 映射目标表
TARGET_TABLE = "jdbc.DATAWAREHOUSE.product_inventory_analysis"

print(f"正在准备写入数据至: {TARGET_TABLE} ...")

try:
    df_result.write \
        .format("adb") \
        .mode("append") \
        .saveAsTable(TARGET_TABLE)
    print(">>> ETL 任务执行完成！数据已成功存入仓库。")
except Exception as e:
    print(f">>> ETL 任务失败，报错原因: {str(e)}")