資料工程師的資料建模

customer ──1────N── order ──1────N── order_line ──N────1── product
                       │
                       N
                       │
                       1
                    address

order_id | order_date | customer_id | customer_city | product_id | product_name | qty | tags
1001     | 2026-11-01 | C1          | Taipei        | P1         | Mug          | 2   | "gift,red"
1001     | 2026-11-01 | C1          | Taipei        | P2         | Pen          | 1   | "office"

-- semi-additive: month-end balance per branch, not the sum of daily balances
SELECT d.year_month,
       b.branch_name,
       SUM(f.balance_amount) AS month_end_balance
FROM   fact_account_balance_daily f
JOIN   dim_date   d ON d.date_key   = f.date_key
JOIN   dim_branch b ON b.branch_key = f.branch_key
WHERE  d.is_month_end
GROUP  BY d.year_month, b.branch_name;

            dim_date                          dim_date
               │                                 │
dim_customer ─ fact_sales ─ dim_product    dim_customer ─ fact_sales ─ dim_product
               │                                 │                         │
            dim_store                         dim_store              dim_subcategory
                                                 │                         │
           (star)                             dim_city                dim_category
                                                       (snowflake)

                      Date  Customer  Product  Store  Promotion  Carrier
Orders                 X       X         X       X        X
Shipments              X       X         X                         X
Returns                X       X         X       X
Inventory snapshot     X                 X       X
Web sessions           X       X

-- drill-across: return rate per product category per month
WITH o AS (
  SELECT d.year_month, p.category, SUM(f.quantity) AS units_sold
  FROM   fact_order_line f
  JOIN   dim_date d    ON d.date_key = f.order_date_key
  JOIN   dim_product p ON p.product_key = f.product_key
  GROUP  BY d.year_month, p.category
), r AS (
  SELECT d.year_month, p.category, SUM(f.quantity) AS units_returned
  FROM   fact_return f
  JOIN   dim_date d    ON d.date_key = f.return_date_key
  JOIN   dim_product p ON p.product_key = f.product_key
  GROUP  BY d.year_month, p.category
)
SELECT o.year_month,
       o.category,
       o.units_sold,
       COALESCE(r.units_returned, 0)                                   AS units_returned,
       ROUND(COALESCE(r.units_returned, 0)::numeric / NULLIF(o.units_sold, 0), 4) AS return_rate
FROM   o
LEFT JOIN r ON r.year_month = o.year_month AND r.category = o.category;

CREATE TABLE fact_order_fulfillment (
  order_id             BIGINT       PRIMARY KEY,   -- degenerate dimension
  customer_key         BIGINT       NOT NULL,
  placed_date_key      INT          NOT NULL,
  paid_date_key        INT,                        -- NULL until the milestone happens
  shipped_date_key     INT,
  delivered_date_key   INT,
  order_amount         NUMERIC(12, 2) NOT NULL,
  days_placed_to_paid  INT,
  days_paid_to_ship    INT,
  days_ship_to_deliver INT
);

-- a shipment event arrives: move the row along
UPDATE fact_order_fulfillment f
SET    shipped_date_key  = sd.date_key,
       days_paid_to_ship = s.ship_date - pd.full_date      -- DATE - DATE = days
FROM   stg_shipments s
JOIN   dim_date sd ON sd.full_date = s.ship_date,
       dim_date pd
WHERE  f.order_id = s.order_id
  AND  pd.date_key = f.paid_date_key
  AND  f.shipped_date_key IS NULL;

CREATE TABLE dim_date (
  date_key      INT PRIMARY KEY,          -- 20261105, readable and sortable
  full_date     DATE NOT NULL UNIQUE,
  year          INT  NOT NULL,
  quarter       INT  NOT NULL,
  month         INT  NOT NULL,
  year_month    CHAR(7) NOT NULL,         -- '2026-11'
  iso_weekday   INT  NOT NULL,            -- 1 = Monday
  is_weekend    BOOLEAN NOT NULL,
  is_month_end  BOOLEAN NOT NULL
);

INSERT INTO dim_date
SELECT CAST(TO_CHAR(g.d, 'YYYYMMDD') AS INT),
       g.d::date,
       EXTRACT(YEAR    FROM g.d)::int,
       EXTRACT(QUARTER FROM g.d)::int,
       EXTRACT(MONTH   FROM g.d)::int,
       TO_CHAR(g.d, 'YYYY-MM'),
       EXTRACT(ISODOW  FROM g.d)::int,
       EXTRACT(ISODOW  FROM g.d) IN (6, 7),
       g.d::date = (date_trunc('month', g.d) + INTERVAL '1 month - 1 day')::date
FROM   generate_series(DATE '2020-01-01', DATE '2030-12-31', INTERVAL '1 day') AS g(d);

CREATE TABLE dim_order_profile (
  order_profile_key  INT PRIMARY KEY,
  is_gift            BOOLEAN NOT NULL,
  payment_type       TEXT    NOT NULL,   -- 'card' | 'wallet' | 'cod'
  order_channel      TEXT    NOT NULL,   -- 'web' | 'ios' | 'android'
  UNIQUE (is_gift, payment_type, order_channel)
);

SELECT od.year_month           AS order_month,
       sd.year_month           AS ship_month,
       COUNT(*)                AS orders
FROM   fact_order_fulfillment f
JOIN   dim_date od ON od.date_key = f.placed_date_key
JOIN   dim_date sd ON sd.date_key = f.shipped_date_key
GROUP  BY od.year_month, sd.year_month;

fact_order_line
  customer_key      ──► dim_customer               (slow attributes: name, signup date)
  demographics_key  ──► dim_customer_demographics  (age_band, income_band, loyalty_tier
                                                    as they were when the order happened)

fact_account_balance ──► dim_account_group ◄── bridge_account_customer ──► dim_customer
                          (account_group_key)    (account_group_key,
                                                  customer_key,
                                                  weight)       -- weights per group sum to 1

-- balance per customer city without double-counting joint accounts
SELECT c.city,
       SUM(f.balance_amount * b.weight) AS weighted_balance
FROM   fact_account_balance f
JOIN   bridge_account_customer b ON b.account_group_key = f.account_group_key
JOIN   dim_customer c            ON c.customer_key      = b.customer_key
WHERE  f.date_key = 20261031
GROUP  BY c.city;

-- every employee under a given manager, any depth (parent-child, recursive CTE)
WITH RECURSIVE reports AS (
  SELECT employee_key, manager_key, 1 AS depth
  FROM   dim_employee
  WHERE  manager_key = 42
  UNION ALL
  SELECT e.employee_key, e.manager_key, r.depth + 1
  FROM   dim_employee e
  JOIN   reports r ON e.manager_key = r.employee_key
)
SELECT * FROM reports;

-- PostgreSQL. dim_customer(customer_sk identity PK, customer_id, address, segment,
--                          valid_from, valid_to, is_current)
BEGIN;

UPDATE dim_customer d
SET    valid_to   = CURRENT_DATE - 1,
       is_current = FALSE
FROM   stg_customer s
WHERE  d.customer_id = s.customer_id
  AND  d.is_current
  AND  (d.address, d.segment) IS DISTINCT FROM (s.address, s.segment);  -- NULL-safe

INSERT INTO dim_customer (customer_id, address, segment, valid_from, valid_to, is_current)
SELECT s.customer_id, s.address, s.segment, CURRENT_DATE, DATE '9999-12-31', TRUE
FROM   stg_customer s
LEFT JOIN dim_customer d
       ON d.customer_id = s.customer_id AND d.is_current
WHERE  d.customer_id IS NULL;   -- brand-new customers, and the ones closed just above

COMMIT;

-- Spark SQL on Delta Lake. <=> is Spark's NULL-safe equality.
-- dim_customer.customer_sk is GENERATED ALWAYS AS IDENTITY.
MERGE INTO dim_customer AS t
USING (
  SELECT s.customer_id AS merge_key, s.*
  FROM   stg_customer s
  UNION ALL
  SELECT CAST(NULL AS STRING) AS merge_key, s.*      -- the new version of a changed key
  FROM   stg_customer s
  JOIN   dim_customer d
         ON d.customer_id = s.customer_id AND d.is_current
  WHERE  NOT (d.address <=> s.address AND d.segment <=> s.segment)
) AS u
ON  t.customer_id = u.merge_key AND t.is_current
WHEN MATCHED AND NOT (t.address <=> u.address AND t.segment <=> u.segment) THEN
  UPDATE SET valid_to = date_sub(current_date(), 1), is_current = false
WHEN NOT MATCHED THEN
  INSERT (customer_id, address, segment, valid_from, valid_to, is_current)
  VALUES (u.customer_id, u.address, u.segment, current_date(), DATE '9999-12-31', true);

-- at fact load time: pick the version valid on the order date
INSERT INTO fact_order_line (order_line_id, order_id, order_date_key, customer_sk, product_key, quantity, net_amount)
SELECT o.order_line_id,
       o.order_id,
       CAST(TO_CHAR(o.order_date, 'YYYYMMDD') AS INT),
       COALESCE(c.customer_sk, -1),                 -- unknown member, never NULL
       COALESCE(p.product_key, -1),
       o.quantity,
       o.net_amount
FROM   stg_order_lines o
LEFT JOIN dim_customer c
       ON c.customer_id = o.customer_id
      AND o.order_date BETWEEN c.valid_from AND c.valid_to
LEFT JOIN dim_product p
       ON p.product_id = o.product_id AND p.is_current;

Kimball (bottom-up)                         Inmon (top-down)
sources ─► staging ─► conformed stars       sources ─► staging ─► 3NF EDW ─► dimensional marts
                       = the warehouse                              (integrated)  (per department)

-- PostgreSQL. Hash keys are MD5 of the normalised business key (32 hex chars).
CREATE TABLE hub_customer (
  hub_customer_hk  CHAR(32)    PRIMARY KEY,
  customer_bk      TEXT        NOT NULL,          -- business key from the source
  load_dts         TIMESTAMP   NOT NULL,
  record_source    TEXT        NOT NULL
);

CREATE TABLE hub_order (
  hub_order_hk     CHAR(32)    PRIMARY KEY,
  order_bk         TEXT        NOT NULL,
  load_dts         TIMESTAMP   NOT NULL,
  record_source    TEXT        NOT NULL
);

CREATE TABLE link_order_customer (
  link_order_customer_hk  CHAR(32)  PRIMARY KEY,  -- hash of both business keys
  hub_order_hk            CHAR(32)  NOT NULL REFERENCES hub_order,
  hub_customer_hk         CHAR(32)  NOT NULL REFERENCES hub_customer,
  load_dts                TIMESTAMP NOT NULL,
  record_source           TEXT      NOT NULL
);

CREATE TABLE sat_customer_details (
  hub_customer_hk  CHAR(32)    NOT NULL REFERENCES hub_customer,
  load_dts         TIMESTAMP   NOT NULL,
  hash_diff        CHAR(32)    NOT NULL,          -- hash of the attributes, to detect change
  name             TEXT,
  email            TEXT,
  address          TEXT,
  record_source    TEXT        NOT NULL,
  PRIMARY KEY (hub_customer_hk, load_dts)
);

models/
  staging/            stg_<source>__<table>.sql    1:1 with a source table: rename, cast, dedupe
  intermediate/       int_<verb>_<entity>.sql      joins and business logic, not exposed
  marts/
    core/             dim_customer.sql, fct_order_line.sql
    finance/          fct_revenue_daily.sql
snapshots/            customers_snapshot.sql        SCD2 of a mutable source table

-- snapshots/customers_snapshot.sql — dbt builds and maintains the SCD2 table
{% snapshot customers_snapshot %}
{{ config(
    target_schema = 'snapshots',
    unique_key    = 'customer_id',
    strategy      = 'timestamp',
    updated_at    = 'updated_at'
) }}
select * from {{ source('app', 'customers') }}
{% endsnapshot %}

# MetricFlow-style definition (the spec has changed across dbt versions — check yours)
semantic_models:
  - name: order_lines
    model: ref('fct_order_line')
    defaults:
      agg_time_dimension: order_date
    entities:
      - name: order_line
        type: primary
        expr: order_line_id
      - name: customer
        type: foreign
        expr: customer_sk
    dimensions:
      - name: order_date
        type: time
        type_params:
          time_granularity: day
    measures:
      - name: net_revenue
        agg: sum
        expr: net_amount

metrics:
  - name: net_revenue
    label: Net revenue
    type: simple
    type_params:
      measure: net_revenue

Process          Grain                                   Date  Customer  Product  Channel  Promo
Orders           one row per order line                   X       X         X        X       X
Returns          one row per returned order line          X       X         X        X
Web sessions     one row per session                      X       X                  X
Inventory        one row per product per warehouse per day X                X

CREATE TABLE dim_customer (
  customer_sk     BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  customer_id     TEXT    NOT NULL,                 -- natural key
  segment         TEXT    NOT NULL,                 -- SCD2-tracked
  country         TEXT    NOT NULL,                 -- SCD2-tracked
  signup_date     DATE    NOT NULL,                 -- type 0
  valid_from      DATE    NOT NULL,
  valid_to        DATE    NOT NULL,
  is_current      BOOLEAN NOT NULL
);

CREATE TABLE dim_product (
  product_key     BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  product_id      TEXT    NOT NULL,
  product_name    TEXT    NOT NULL,
  subcategory     TEXT    NOT NULL,                 -- flattened fixed-depth hierarchy
  category        TEXT    NOT NULL,
  brand           TEXT    NOT NULL,
  is_current      BOOLEAN NOT NULL
);

CREATE TABLE fact_order_line (
  order_line_id      BIGINT       PRIMARY KEY,
  order_id           BIGINT       NOT NULL,         -- degenerate dimension
  order_date_key     INT          NOT NULL REFERENCES dim_date (date_key),
  customer_sk        BIGINT       NOT NULL REFERENCES dim_customer (customer_sk),
  product_key        BIGINT       NOT NULL REFERENCES dim_product (product_key),
  channel_key        INT          NOT NULL,
  promotion_key      INT          NOT NULL,         -- -2 = "no promotion"
  quantity           INT          NOT NULL,
  gross_amount       NUMERIC(12, 2) NOT NULL,
  discount_amount    NUMERIC(12, 2) NOT NULL,
  net_amount         NUMERIC(12, 2) NOT NULL        -- additive
);

-- 1. revenue by category by month
SELECT d.year_month, p.category, SUM(f.net_amount) AS net_revenue, SUM(f.quantity) AS units
FROM   fact_order_line f
JOIN   dim_date d    ON d.date_key    = f.order_date_key
JOIN   dim_product p ON p.product_key = f.product_key
GROUP  BY d.year_month, p.category;

-- 4a. revenue by segment as it was when each order happened
SELECT c.segment, SUM(f.net_amount) AS net_revenue
FROM   fact_order_line f
JOIN   dim_customer c ON c.customer_sk = f.customer_sk
GROUP  BY c.segment;

-- 4b. revenue by segment as it is today
SELECT cur.segment, SUM(f.net_amount) AS net_revenue
FROM   fact_order_line f
JOIN   dim_customer hist ON hist.customer_sk = f.customer_sk
JOIN   dim_customer cur  ON cur.customer_id  = hist.customer_id AND cur.is_current
GROUP  BY cur.segment;