ODS 与 DWD 实现

本章目标：自己把 8 个 CSV 落成 ODS，再写出两张 DWD 宽表；每完成一步都用行数校验，数字不对先别往下走。 前置知识：CREATE TABLE AS SELECT、JOIN、GROUP BY、CASE WHEN、Hive 分区 预计用时：90 分钟。

0. 本章你要产出什么

在实验室库 project/warehouse_lab.duckdb 里建出：

完成标志：下面这条查询得到 4 个正确数字。

SELECT (SELECT COUNT(*) FROM ods_orders)        AS ods_orders,
       (SELECT COUNT(*) FROM dwd_order_detail)  AS detail_rows,
       (SELECT COUNT(DISTINCT order_id) FROM dwd_order_detail) AS detail_orders,
       (SELECT COUNT(*) FROM dwd_order)         AS valid_orders;

1. 先看懂：ODS 和 DWD 的边界

ODS 的职责是 原样保存。源系统给什么就存什么，字段名不改、脏数据不删、业务规则不加。这样一旦下游口径出错，可以回到 ODS 重跑，而不是回头找业务系统要数据。

DWD 的职责是 把数据变成可分析的干净明细：

• 统一时间类型；
• 排除无效订单；
• 关联维度，补上可读字段；
• 控制粒度，避免 fan-out；
• 保留原始异常，只做标记，不轻易删除。

ODS 是“证据”，DWD 是“可用的明细”。两层不能合并。

2. 动手一：落 8 张 ODS 表

2.1 先建一张，确认路径和类型

先确认数据已经下载：按 00 章第 4 节 的链接下载 zip，解压到 data/raw/olist/，然后跑体检：

python scripts/check_data.py

应看到 8 张表 / 550,759 行。还没下载就先回 01 章，不要跳过这一步。

然后在实验室里建第一张表。把下面的 SQL 存成任意文件，或直接在 project/run_sql.py 的交互模式里执行：

CREATE OR REPLACE TABLE ods_orders AS
SELECT * FROM read_csv_auto('data/raw/olist/olist_orders_dataset.csv', header = true);

执行方式：

python project/run_sql.py -c "CREATE OR REPLACE TABLE ods_orders AS SELECT * FROM read_csv_auto('data/raw/olist/olist_orders_dataset.csv', header = true)"
python project/run_sql.py -c "SELECT COUNT(*) FROM ods_orders"

应该看到 99441。如果报“文件不存在”，先回到项目根目录，再检查 data/raw/olist/。

DuckDB 自动推断类型：read_csv_auto 会把时间列推断成 TIMESTAMP、金额列推断成 DOUBLE。ODS 不手工改字段名、不删行、不做业务过滤；DWD 里仍然显式 CAST，这样换引擎或换文件格式时行为一致。

2.2 把 8 张表一次建完

照着上面的写法，把 8 个文件都落成 ods_* 表。自己先写一遍，再对照答案：

python project/run_sql.py project/lab/01_ods.sql

2.3 校验：8 张表、550,759 行

SELECT 'orders' AS table_name, COUNT(*) AS row_cnt FROM ods_orders
UNION ALL SELECT 'order_items', COUNT(*) FROM ods_order_items
UNION ALL SELECT 'payments', COUNT(*) FROM ods_payments
UNION ALL SELECT 'customers', COUNT(*) FROM ods_customers
UNION ALL SELECT 'products', COUNT(*) FROM ods_products
UNION ALL SELECT 'sellers', COUNT(*) FROM ods_sellers
UNION ALL SELECT 'order_reviews', COUNT(*) FROM ods_order_reviews
UNION ALL SELECT 'category_translation', COUNT(*) FROM ods_category_translation
ORDER BY table_name;

结果如下：

合计 550,759 行。这 8 个数对不上时，先检查是不是漏了文件、或者把表名写串了。

2.4 八张表的粒度与主键

注意：ods_customers 一行是一个收货地址，不是一个人。自然人 ID 是 customer_unique_id，做用户级指标时别用错。

2.5 为什么 ODS 不手工清洗

• 原始值可追溯：下游算错时，能回到原始列重算；
• 规则可回滚：清洗口径改了，只需重跑 DWD，不用重新下载；
• 职责清晰：不同下游对脏数据的处理方式不同，ODS 不替它们做决定。

真实生产会用 DataX / Flink CDC 把数据采集到 HDFS/对象存储，再建外部表（Sqoop 已进入 Apache Attic，老系统仍可能见到）：

CREATE EXTERNAL TABLE ods_orders (
    order_id STRING,
    customer_id STRING,
    order_status STRING,
    order_purchase_timestamp STRING,
    order_approved_at STRING,
    order_delivered_carrier_date STRING,
    order_delivered_customer_date STRING,
    order_estimated_delivery_date STRING
)
PARTITIONED BY (dt STRING)
STORED AS PARQUET
LOCATION 'hdfs:///warehouse/ods/olist/orders';

ODS 用外部表，避免 DROP TABLE 时把原始数据一起删掉。

3. 动手二：先体检，再写清洗规则

清洗规则不能拍脑袋。先跑几条体检 SQL，看清楚空值、重复、一对多和时间异常，再决定 DWD 里做什么。

3.1 空值：多数是业务正常

SELECT
    COUNT(*) FILTER (WHERE order_approved_at IS NULL)              AS approved_null,
    COUNT(*) FILTER (WHERE order_delivered_carrier_date IS NULL)   AS carrier_null,
    COUNT(*) FILTER (WHERE order_delivered_customer_date IS NULL)  AS delivered_null,
    COUNT(*) FILTER (
        WHERE order_status = 'delivered'
          AND order_delivered_customer_date IS NULL
    ) AS delivered_status_missing_date,
    COUNT(*) FILTER (
        WHERE order_status NOT IN ('canceled', 'unavailable')
          AND order_delivered_customer_date IS NULL
    ) AS valid_missing_date
FROM ods_orders;

关键点：空值不一定是脏数据。未发货订单没有签收时间，是业务事实；只有“状态已签收却没有签收时间”才是异常，共 8 条。

还有两类空值：

3.2 重复与一对多

SELECT
    (SELECT COUNT(*) FROM (
        SELECT order_id FROM ods_orders GROUP BY 1 HAVING COUNT(*) > 1
    )) AS duplicate_orders,
    (SELECT COUNT(*) FROM (
        SELECT order_id, order_item_id FROM ods_order_items
        GROUP BY 1, 2 HAVING COUNT(*) > 1
    )) AS duplicate_items,
    (SELECT COUNT(*) FROM (
        SELECT order_id, payment_sequential FROM ods_payments
        GROUP BY 1, 2 HAVING COUNT(*) > 1
    )) AS duplicate_payments,
    (SELECT COUNT(*) FROM (
        SELECT review_id FROM ods_order_reviews GROUP BY 1 HAVING COUNT(*) > 1
    )) AS duplicate_review_ids,
    (SELECT COUNT(*) - COUNT(DISTINCT review_id) FROM ods_order_reviews) AS extra_review_rows,
    (SELECT COUNT(*) FROM (
        SELECT order_id FROM ods_order_reviews GROUP BY 1 HAVING COUNT(*) > 1
    )) AS multi_review_orders;

前 3 个主键干净，不需要去重。评价表例外：review_id 会重复，而且同一单可能有多条评价。不能简单 DISTINCT，也不能直接 JOIN 明细，正确做法是先按 order_id 聚合均分。

3.3 没有明细的订单

SELECT
    (SELECT COUNT(*) FROM ods_orders o
     WHERE NOT EXISTS (
         SELECT 1 FROM ods_order_items i WHERE i.order_id = o.order_id
     )) AS all_orders_without_items,
    (SELECT COUNT(*) FROM ods_orders o
     WHERE o.order_status NOT IN ('canceled', 'unavailable')
       AND NOT EXISTS (
           SELECT 1 FROM ods_order_items i WHERE i.order_id = o.order_id
       )) AS valid_orders_without_items;

结果 775 / 8。全部订单里 775 单没有明细；只看有效订单，还有 8 单。所以：

• 有效订单 = 98,207；
• 有明细的有效订单 = 98,199；
• AOV 的分母只能用 98,199。

3.4 时间倒挂

SELECT
    COUNT(*) FILTER (
        WHERE order_delivered_carrier_date < order_purchase_timestamp
    ) AS carrier_before_purchase,
    COUNT(*) FILTER (
        WHERE order_delivered_customer_date < order_delivered_carrier_date
    ) AS delivered_before_carrier,
    COUNT(*) FILTER (
        WHERE order_approved_at < order_purchase_timestamp
    ) AS approved_before_purchase
FROM ods_orders;

这些行不删除，质量章单独列出。删除会让订单数、GMV 对不上源系统。

3.5 价格字段：先看分布再决定过滤

SELECT
    COUNT(*) FILTER (WHERE price < 0)          AS negative_price,
    COUNT(*) FILTER (WHERE price = 0)          AS zero_price,
    COUNT(*) FILTER (WHERE freight_value = 0)  AS free_shipping,
    MIN(price) AS min_price,
    MAX(price) AS max_price
FROM ods_order_items;

所以 DWD 不需要过滤 price > 0。清洗规则要按真实数据分布制定，照搬通用模板容易多删数据。

3.6 体检结论 → DWD 只做六件事

1. 排除 canceled 和 unavailable；
2. 把时间列显式 CAST 成 DATE / TIMESTAMP；
3. 关联客户、商品、品类翻译、卖家；
4. 派生 quantity = 1、item_amount、item_amount_with_freight；
5. 评价先按订单聚合，再 JOIN；
6. 客户累计消费分档派生 customer_level。

支付与明细的差异不在 DWD 处理，留到数据质量对账。

4. 动手三：DWD 明细宽表

4.1 第一步：只做过滤和关联

先不要一次写完。第一个版本只保留明细、关联订单、排除无效状态：

CREATE OR REPLACE TABLE dwd_order_detail AS
SELECT
    i.order_id,
    CAST(i.order_item_id AS INTEGER) AS item_seq,
    CAST(o.order_purchase_timestamp AS DATE) AS order_date,
    o.order_status,
    i.product_id,
    i.seller_id,
    CAST(i.price AS DECIMAL(12, 2)) AS price,
    CAST(i.freight_value AS DECIMAL(12, 2)) AS freight
FROM ods_order_items i
JOIN ods_orders o USING (order_id)
WHERE o.order_status NOT IN ('canceled', 'unavailable');

自查：

SELECT COUNT(*) AS detail_rows, COUNT(DISTINCT order_id) AS detail_orders
FROM dwd_order_detail;

应该得到 112,101 / 98,199。如果错了，先检查 WHERE 的两个状态和 JOIN 条件。

4.2 第二步：补维度、金额和评价

在第一步的基础上，加客户、品类、卖家维度和评价聚合：

CREATE OR REPLACE TABLE dwd_order_detail AS
WITH review AS (
    SELECT order_id, AVG(CAST(review_score AS INTEGER)) AS review_score
    FROM ods_order_reviews
    GROUP BY order_id
)
SELECT
    i.order_id,
    CAST(i.order_item_id AS INTEGER) AS item_seq,
    CAST(o.order_purchase_timestamp AS DATE) AS order_date,
    o.order_status,
    o.customer_id,
    c.customer_unique_id,
    c.customer_city,
    c.customer_state,
    i.product_id,
    COALESCE(t.product_category_name_english, 'unknown') AS category,
    i.seller_id,
    s.seller_state,
    CAST(i.price AS DECIMAL(12, 2)) AS price,
    1 AS quantity,
    CAST(i.freight_value AS DECIMAL(12, 2)) AS freight,
    CAST(i.price AS DECIMAL(14, 2)) AS item_amount,
    CAST(CAST(i.price AS DECIMAL(14, 2))
         + CAST(i.freight_value AS DECIMAL(14, 2)) AS DECIMAL(14, 2))
        AS item_amount_with_freight,
    r.review_score
FROM ods_order_items i
JOIN ods_orders o USING (order_id)
LEFT JOIN ods_customers c USING (customer_id)
LEFT JOIN ods_products p USING (product_id)
LEFT JOIN ods_category_translation t USING (product_category_name)
LEFT JOIN ods_sellers s USING (seller_id)
LEFT JOIN review r USING (order_id)
WHERE o.order_status NOT IN ('canceled', 'unavailable');

再自查一次，行数必须仍然是 112,101。如果 JOIN 后变多，说明有关联键不是 1:1，或者评价没有先聚合。

python project/run_sql.py project/sql/dwd_order_detail.sql

4.3 校验：行数、订单数和金额

SELECT COUNT(*)                              AS detail_rows,
       COUNT(DISTINCT order_id)              AS detail_orders,
       ROUND(SUM(item_amount_with_freight), 2) AS gmv
FROM dwd_order_detail;

结果：

GMV 能提前对上，说明明细金额没有重复计算。

4.4 行数漏斗

ods_orders                 99,441
  - canceled 625
  - unavailable 609
= dwd_order                98,207

ods_order_items           112,650
  ∩ 有效订单
= dwd_order_detail        112,101

dwd_order_detail 去重订单    98,199（8 个有效订单没有明细）

这四个数就是本章最重要的验收线：

• 99,441 是全部订单；
• 98,207 是排除了 625 + 609 个无效订单后的有效订单；
• 112,101 是这些有效订单关联到的明细行；
• 98,199 是有明细的有效订单，比 98,207 少 8 个。

5. 动手四：DWD 订单宽表

5.1 先建订单骨架

明细表是“订单 × 商品行”，很多指标天然按订单算。先建一张只有订单字段的表：

CREATE OR REPLACE TABLE dwd_order AS
SELECT
    o.order_id,
    o.customer_id,
    o.order_status,
    CAST(o.order_purchase_timestamp AS DATE) AS order_date
FROM ods_orders o
WHERE o.order_status NOT IN ('canceled', 'unavailable');

自查：

SELECT COUNT(*) AS valid_orders FROM dwd_order;

应该是 98,207。这张表必须从 ods_orders 出发，不能从 dwd_order_detail 出发，否则会丢掉 8 个没有明细的有效订单。

5.2 补金额、评价和履约字段

订单级金额先聚合明细，再加上客户、评价和履约时间。核心派生字段：

CASE WHEN delivered_at IS NOT NULL
      AND estimated_at IS NOT NULL
      AND delivered_at > estimated_at
     THEN TRUE ELSE FALSE END AS is_late,
CASE WHEN delivered_at IS NOT NULL
     THEN date_diff('day', order_date, CAST(delivered_at AS DATE))
END AS delivery_days

组装方式：

1. item CTE：按 order_id 聚合 item_cnt、商品金额、运费、订单总额；
2. review CTE：按 order_id 聚合评价均分；
3. base：从 ods_orders 出发，LEFT JOIN 上面两个 CTE 和客户表；
4. 最后派生 is_late 和 delivery_days。

python project/run_sql.py project/sql/dwd_order.sql

5.3 校验：订单、空金额和履约

SELECT COUNT(*)                                        AS valid_orders,
       COUNT(*) FILTER (WHERE order_amount IS NULL)    AS no_item_orders,
       COUNT(*) FILTER (WHERE delivered_at IS NULL)    AS no_delivery_time,
       COUNT(*) FILTER (
           WHERE delivered_at IS NOT NULL AND estimated_at IS NOT NULL
       )                                               AS late_base,
       COUNT(*) FILTER (WHERE is_late)                 AS late_orders
FROM dwd_order;

结果：

no_item_orders = 8 不是错误，是订单生命周期分析需要的保留行。它们不进 GMV，也不进 AOV 分母。

6. 为什么这样建模

事实表 dwd_order_detail 的粒度是 订单 × 商品行。维度退化字段直接放在事实表：

没有单独建 dim_date、dim_customer 表，是因为数据量小、分析场景固定。真实数仓会把稳定维度拆出去，做缓慢变化维（SCD）。项目在 SCD 章节 单独讲了拉链表，但没有强行套到本项目。

6.1 星型模型

dim_customer ──┐
dim_product  ──┼── fact_order_item ── dim_seller
dim_date     ──┘

星型模型的核心是：一张事实表，周围一圈维度表。事实表存度量和外键，维度表存描述属性。不要把所有字段塞进一张大宽表，也不要把一张表拆成十几张。

6.2 为什么评价和支付不能直接 JOIN 明细

payments 一单可能多行，order_reviews 一单也可能多条。直接 JOIN 明细会让行数膨胀，GMV 和评价均分都会被重复计算。本章的做法是：评价先 GROUP BY order_id 求均分，再 LEFT JOIN；支付只做质量对账，不进入 GMV 主链路。

7. 可选：Hive 动态分区写入

本地必修路径是 DuckDB。想练 Hive 写法，可以看 project/hive/hive_ddl.sql：

SET hive.exec.dynamic.partition = true;
SET hive.exec.dynamic.partition.mode = nonstrict;

INSERT OVERWRITE TABLE dw_demo.dwd_order_detail PARTITION (dt)
SELECT
    i.order_id,
    i.order_item_id AS item_seq,
    substr(o.order_purchase_timestamp, 1, 10) AS order_date,
    o.order_status,
    o.customer_id,
    c.customer_unique_id,
    i.product_id,
    COALESCE(t.product_category_name_english, 'unknown') AS category,
    i.price,
    1 AS quantity,
    i.freight_value AS freight,
    i.price AS item_amount,
    i.price + i.freight_value AS item_amount_with_freight,
    substr(o.order_purchase_timestamp, 1, 10) AS dt
FROM dw_demo.ods_order_items i
JOIN dw_demo.ods_orders o ON i.order_id = o.order_id
LEFT JOIN dw_demo.ods_customers c ON o.customer_id = c.customer_id
LEFT JOIN dw_demo.ods_products p ON i.product_id = p.product_id
LEFT JOIN dw_demo.ods_category_translation t
  ON p.product_category_name = t.product_category_name
WHERE o.order_status NOT IN ('canceled', 'unavailable');

动态分区列 dt 必须放 SELECT 最后一列。脚本未在真实集群验证，只作为写法示例。

8. 本章验收

• 能说出 8 张 ODS 表的主键和粒度。
• 8 张 ODS 表行数全对，合计 550,759。
• 能解释 99,441 → 98,207 → 112,101 → 98,199 每一步为什么变。
• 能说出至少三类真实脏数据，并区分“业务正常空值”和“异常空值”。
• 能写出评价先聚合再 JOIN 的 SQL，并解释 fan-out。
• 能解释为什么 Olist 不需要过滤负价。
• 能说明 8 个无明细订单保留在 dwd_order 的原因。
• 实验室里的 dwd_order_detail 是 112,101 行、98,199 个订单，dwd_order 是 98,207 行。

9. 面试延伸

面试题：ODS 为什么原样保存，不直接清洗？

是什么：ODS 是贴源层，DWD 是清洗层。

解决什么问题：原样保存让数据可追溯、可重跑。如果 ODS 就清洗，口径一改就无法回滚；而且不同下游对脏数据的处理方式不同，ODS 不应替它们做决定。

我的实操边界：项目里 ODS 保持 CSV 原始字段，DWD 才做类型转换和过滤。真实生产还要考虑采集幂等、分区覆盖和 schema 演进。

面试题：为什么要单独建订单宽表 dwd_order？

是什么：明细表是订单 × 商品行，订单宽表是一行一个订单。

解决什么问题：订单数、AOV、复购、履约这些指标天然是订单粒度。在明细表上 COUNT(DISTINCT order_id) 也能算，但每次都去重，性能差且容易写错。订单宽表把订单级金额和履约字段预聚合好，下游更稳定。

我的实操边界：项目里 dwd_order 有 98,207 行，包含 8 个无明细订单；金额为空的订单不会被算进 AOV。

面试题：fan-out 是什么，怎么避免？

是什么：一对多 JOIN 导致事实行数膨胀，指标被重复计算。

解决什么问题：支付、评价、优惠券这类“一单多行”的表，先聚合到订单粒度，再 JOIN 明细；或者用 EXISTS / 标量子查询取值。

我的实操边界：项目里评价先 GROUP BY order_id 求均分，再 LEFT JOIN；支付只做质量对账，不进 GMV 主链路。