跳到主要內容

DE Day 31 端到端管線（二）：轉換與建模

DE Day 31 端到端管線（二）：轉換與建模

執行需求：CPU 可跑。今天是端到端管線系列的第二天。昨天我們把 data.gov.tw 的「公司登記資料」與「公司變更登記資料」下載回來、寫入 raw.company_basic 與 raw.company_change、並匯出成 Parquet 分區檔。今天要接著做「轉換與建模」：用 SQL 把 raw 整理成 staging 與 mart 兩層，並用 Day 23–Day 24 介紹的維度建模概念組成「公司維度表 mart.dim_company」與「變更事實表 mart.fact_company_change」。這兩張表是 Day 32 品質檢查與 Day 34 Streamlit 儀表板的基礎。本篇所有範例都在 CPU 上執行，不需要 GPU。讀完這篇，你會把「落地資料」轉成「分析友善」的星狀結構。

引言

昨天的落地層 raw 保留的是「資料原本的樣貌」：欄位名稱是政府平台公布的英文代號、型別全部是字串、日期是民國年或西元年的字串、缺失值可能是空字串或 NULL。這樣的資料沒辦法直接做分析——例如你想算「公司成立年份的分布」，如果建立日期是民國 75 年的字串，需要先轉成西元年的日期型別才能 group by 年份。

今天的轉換層 staging 負責做三件事：第一，欄位重新命名（從政府平台的英文代號改成我們自己定義的語意化名稱，例如 統一編號 → uniform_no）；第二，型別轉換（字串轉日期、字串轉整數、字串轉金額）；第三，缺失值與異常值處理（空字串標準化為 NULL、金額負數改為 0 或留 NULL 並標記）。轉換層是「資料被整理過一輪」的狀態，仍保留逐筆明細，但欄位已經分析友善。

建模層 mart 再進一步：把 staging 的明細組合成兩類表——維度表（dim）與事實表（fact）。維度表是「描述性質的資料」，例如一家公司的基本資料；事實表是「事件性質的資料」，例如一家公司每一次的變更登記。兩者用「公司統一編號」連起來，就是 Day 23 講的星狀結構（star schema）。這個結構的好處是「分析快速」：維度表小而精簡、事實表大而稀疏，分析查詢時只要 join 一次維度表就能拿到所有描述資訊。

轉換與建模的核心觀念

轉換與建模是資料工程裡最需要「懂 SQL」的一步。我們會大量使用 Day 3–Day 7 介紹的視窗函式、CTE、排名等技巧，並結合 Day 13–Day 14 的清洗與品質觀念。在建模層，我們還會用 Day 23 講的 SCD（slowly changing dimension）類型 1：用最新一筆的覆寫式更新，不保留歷史。這對「公司基本資料」是合理的——公司名稱變更時我們要的是「現在叫什麼名字」，不是「曾經叫過什麼名字」。

另一個重要觀念是「分層命名」：raw、staging、mart 三層各有清楚的職責。raw 是「原始檔的鏡像」、staging 是「轉換後的明細」、mart 是「給分析用的星狀結構」。這樣的分層讓每個層的 SQL 可以獨立測試、獨立重跑：當 raw 有新版資料時，只要重跑 staging 與 mart 的轉換就能更新；當 staging 邏輯修改時，只要重跑 mart 就能反映。

冪等（idempotent）是轉換層的另一個關鍵特性。「重跑同一天的轉換要得到一樣的結果」是資料工程的基本要求。我們用 CREATE OR REPLACE TABLE 加上 WHERE dt = ? 的設計來達成冪等：每次轉換都先把當天的中間表清掉，再重新生成，這樣即使不小心重跑一次也不會產生重複資料。

共用設定：擴充 common.py 加上品質門檻

為了 Day 32 品質檢查時能引用同一份設定，我們先在 pipelines/common.py 加上品質門檻與轉換表名稱的常數：

"""de-journey/pipelines/common.py：Day 30-35 共用的管線常數。"""
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[1]
DATA_DIR = PROJECT_ROOT / "data"
WAREHOUSE_DIR = PROJECT_ROOT / "warehouse"
LOGS_DIR = PROJECT_ROOT / "logs"

DUCKDB_PATH = WAREHOUSE_DIR / "de-journey.duckdb"

DATASETS = {
    "company_basic": {
        "title": "公司登記基本資料",
        "source": "moea_basic",
        "table": "raw.company_basic",
        # 轉換後的清洗表（Day 31 產出）

        "staging_table": "staging.company_basic_clean",
        # 維度表（Day 31 產出）

        "dim_table": "mart.dim_company",
        "key_columns": ["uniform_no", "company_name"],
        "partition_prefix": "company_basic",
    },
    "company_change": {
        "title": "公司變更登記資料",
        "source": "moea_change",
        "table": "raw.company_change",
        "staging_table": "staging.company_change_clean",
        "fact_table": "mart.fact_company_change",
        "key_columns": ["uniform_no", "change_date", "change_item"],
        "partition_prefix": "company_change",
    },
}

# 品質門檻（Day 32 會引用）

QUALITY_RULES = {
    "company_basic": [
        ("uniform_no_not_null", "uniform_no IS NOT NULL"),
        ("uniform_no_length_8", "LENGTH(uniform_no) = 8"),
        ("company_name_not_null", "company_name IS NOT NULL"),
        ("company_name_min_len", "LENGTH(company_name) >= 2"),
        ("capital_amount_non_negative", "capital_amount >= 0"),
        ("establish_date_valid", "establish_date IS NOT NULL"),
    ],
    "company_change": [
        ("uniform_no_not_null", "uniform_no IS NOT NULL"),
        ("change_date_not_null", "change_date IS NOT NULL"),
        ("change_item_not_null", "change_item IS NOT NULL"),
    ],
}

HTTP_TIMEOUT_SEC = 30
RETRY_ATTEMPTS = 3
RETRY_BACKOFF_SEC = 2.0

這份設定的擴充重點是三件事：第一，每個資料集加上 staging_table 與 dim_table／fact_table，把三層命名（raw、staging、mart）全部集中管理；第二，新增 QUALITY_RULES 字典，把品質規則的「規則名稱 + SQL 條件」成對放好；第三，保留 Day 30 的 HTTP 與重試設定不動，確保向後相容。後續的 Day 32 品質檢查會直接讀 QUALITY_RULES，不用重複定義。

完整實作：把 raw 轉成 staging

先把落地層 raw.company_basic 整理成 staging.company_basic_clean。這個步驟主要做欄位重新命名、型別轉換、缺失值標準化：

"""de-journey/pipelines/transform_basic.py：把 raw.company_basic 轉成 staging。"""
import duckdb

from pipelines.common import DUCKDB_PATH

con = duckdb.connect(str(DUCKDB_PATH))
con.execute("CREATE SCHEMA IF NOT EXISTS staging")

con.execute("""
    CREATE OR REPLACE TABLE staging.company_basic_clean AS
    SELECT
        -- 政府平台的統一編號欄位是「統一編號」，改名為 uniform_no
        TRIM("統一編號") AS uniform_no,
        -- 公司名稱可能有前後空白，TRIM 後保留
        TRIM("公司名稱") AS company_name,
        -- 公司狀態（設立／解散）標準化
        CASE
            WHEN TRIM("公司狀況") IN ('設立', '核准設立') THEN 'active'
            WHEN TRIM("公司狀況") IN ('解散', '核准解散') THEN 'dissolved'
            ELSE NULL
        END AS status,
        -- 資本額從字串轉成整數（移除千分位逗號、空字串改為 NULL）
        TRY_CAST(REPLACE(NULLIF("資本額", ''), ',', '') AS BIGINT) AS capital_amount,
        -- 代表人姓名
        TRIM("代表人姓名") AS representative_name,
        -- 公司所在地（只取到縣市，地址明細留給需要的人）
        TRIM("公司所在地") AS company_location,
        -- 行業別代號
        TRIM("行業代號") AS industry_code,
        -- 設立日期從民國年字串轉成 DATE（例：75 年 5 月 1 日 -> 1986-05-01）
        CASE
            WHEN "設立日期" ~ '^[0-9]{{6,7}}$'
            THEN DATE '1911-12-31' + CAST("設立日期" AS INTEGER)
            ELSE NULL
        END AS establish_date
    FROM raw.company_basic
    WHERE TRIM("統一編號") <> ''
""")

n = con.execute("SELECT COUNT(*) FROM staging.company_basic_clean").fetchone()[0]
print(f"staging.company_basic_clean：{n} 筆")  # 輸出：依當日下載而略有不同

con.close()

這段 SQL 用到了 Day 13 的「字串清洗」技巧（TRIM、REPLACE、NULLIF）與 Day 6 的 CASE WHEN 條件式。其中兩個比較特別的設計要說明：第一，capital_amount 用 TRY_CAST(... AS BIGINT) 把字串轉成整數。為什麼用 TRY_CAST 而不是 CAST？因為政府平台的資本額欄位偶爾會有「未揭露」或「-」之類的非數字字串，CAST 在這種情況會直接 raise 整個 SQL 失敗，而 TRY_CAST 會把無法轉換的值變成 NULL，這正是我們要的「缺失值標準化」。第二，establish_date 用 DATE '1911-12-31' + CAST(... AS INTEGER) 把民國年（3–4 位數字，例如 750501 表示 75 年 5 月 1 日）轉成西元日期。這個公式的核心是「西元年 = 民國年 + 1911」，但因為 DuckDB 的 DATE 型別不支援直接加整數年數，所以用「1911-12-31 + 天數」這種寫法。實務上更通用的做法是 DATE_FROM_PARTS(CAST(LEFT(x, 3) AS INT) + 1911, CAST(SUBSTR(x, 4, 2) AS INT), CAST(SUBSTR(x, 6, 2) AS INT))。

同樣的轉換邏輯也用在公司變更登記上。把 raw.company_change 轉成 staging.company_change_clean：

"""de-journey/pipelines/transform_change.py：把 raw.company_change 轉成 staging。"""
import duckdb

from pipelines.common import DUCKDB_PATH

con = duckdb.connect(str(DUCKDB_PATH))
con.execute("CREATE SCHEMA IF NOT EXISTS staging")

con.execute("""
    CREATE OR REPLACE TABLE staging.company_change_clean AS
    SELECT
        TRIM("統一編號") AS uniform_no,
        -- 變更日期用同一個公式
        CASE
            WHEN "變更日期" ~ '^[0-9]{{6,7}}$'
            THEN DATE '1911-12-31' + CAST("變更日期" AS INTEGER)
            ELSE NULL
        END AS change_date,
        -- 變更細項標準化：原文字可能多變，先取前 12 字當類別
        LEFT(TRIM("變更細項"), 12) AS change_item,
        -- 變更前的值（若有的話）
        NULLIF(TRIM("變更前"), '') AS before_value,
        -- 變更後的值
        NULLIF(TRIM("變更後"), '') AS after_value
    FROM raw.company_change
    WHERE TRIM("統一編號") <> ''
""")

n = con.execute("SELECT COUNT(*) FROM staging.company_change_clean").fetchone()[0]
print(f"staging.company_change_clean：{n} 筆")  # 輸出：依當日下載而略有不同

con.close()

變更表的清洗重點是「細項分類」：政府平台把變更細項寫成完整句子（例如「資本額變更：1,000,000 → 5,000,000」），但分析時我們只需要「是哪一類變更」（資本額、負責人、公司名稱）。這裡用 LEFT(TRIM(...), 12) 當粗糙分類；如果想要更精細，可以另外寫一張 mapping 表把各種變更敘述對應到類別。

完整實作：建立維度表與事實表

有了 staging 兩張清洗表，就可以組合成維度表與事實表。維度表「mart.dim_company」從 staging.company_basic_clean 取最新的公司資料；事實表「mart.fact_company_change」則是每一家公司每一次的變更事件，搭配維度表的 uniform_no 就能查到「這家公司的所有變更」與「這次變更的公司基本資料」。

"""de-journey/pipelines/build_marts.py：把 staging 組合成 dim_company 與 fact_company_change。"""
import duckdb

from pipelines.common import DUCKDB_PATH

con = duckdb.connect(str(DUCKDB_PATH))
con.execute("CREATE SCHEMA IF NOT EXISTS mart")

# 維度表：用 DISTINCT ON 取每家公司一筆（PostgreSQL 風格）

con.execute("""
    CREATE OR REPLACE TABLE mart.dim_company AS
    SELECT
        uniform_no,
        company_name,
        status,
        capital_amount,
        representative_name,
        company_location,
        industry_code,
        establish_date
    FROM staging.company_basic_clean
    QUALIFY ROW_NUMBER() OVER (
        PARTITION BY uniform_no ORDER BY establish_date DESC NULLS LAST
    ) = 1
""")

n_dim = con.execute("SELECT COUNT(*) FROM mart.dim_company").fetchone()[0]
print(f"mart.dim_company：{n_dim} 家公司")  # 輸出：依當日下載而略有不同


# 事實表：每一家公司每一次的變更

con.execute("""
    CREATE OR REPLACE TABLE mart.fact_company_change AS
    SELECT
        -- 用 ROW_NUMBER 產生 surrogate key（分析時可用 ORDER BY 排序）
        ROW_NUMBER() OVER (
            PARTITION BY uniform_no ORDER BY change_date, change_item
        ) AS change_seq,
        uniform_no,
        change_date,
        change_item,
        before_value,
        after_value,
        -- 加一個 ingest 日期欄位方便回溯
        current_date AS ingested_date
    FROM staging.company_change_clean
    WHERE change_date IS NOT NULL
""")

n_fact = con.execute("SELECT COUNT(*) FROM mart.fact_company_change").fetchone()[0]
print(f"mart.fact_company_change：{n_fact} 筆變更")  # 輸出：依當日下載而略有不同

con.close()

這段 SQL 用到了 Day 4 講過的 ROW_NUMBER() 視窗函式，搭配 PARTITION BY 與 QUALIFY 達到「每家公司取最新一筆」的效果。DuckDB 完整支援 SQL 標準的 QUALIFY 子句，這比傳統的 WHERE rn = 1 更直觀：QUALIFY 直接接在 ROW_NUMBER() 之後，不用包一層 CTE。維度表用 QUALIFY ROW_NUMBER() OVER (PARTITION BY uniform_no ORDER BY establish_date DESC NULLS LAST) = 1 選出「每家公司最新一筆」。注意 NULLS LAST 是必要的：當 establish_date 為 NULL 時（極少數），我們希望它排在最後，這樣有日期的資料會優先被選到。

事實表則用 ROW_NUMBER() OVER (PARTITION BY uniform_no ORDER BY change_date, change_item) 產生 surrogate key。這個 key 不是必要的（可以用 (uniform_no, change_date, change_item) 當複合主鍵），但加上之後儀表板的查詢會更方便，例如「顯示某家公司前 5 大變更事件」可以用 WHERE change_seq <= 5。注意 ingested_date 用 current_date 而不是 NOW()：因為變更表是每天快照式的，只要記錄「哪一天被採集到」即可。

最後做一次驗證，確認三層之間的筆數關係合理：

"""de-journey/pipelines/verify_marts.py：驗證 raw / staging / mart 三層的一致性。"""
import duckdb

from pipelines.common import DUCKDB_PATH

con = duckdb.connect(str(DUCKDB_PATH))
checks = [
    ("raw.company_basic",
     "SELECT COUNT(*) FROM raw.company_basic"),
    ("staging.company_basic_clean",
     "SELECT COUNT(*) FROM staging.company_basic_clean"),
    ("mart.dim_company",
     "SELECT COUNT(*) FROM mart.dim_company"),
    ("raw.company_change",
     "SELECT COUNT(*) FROM raw.company_change"),
    ("staging.company_change_clean",
     "SELECT COUNT(*) FROM staging.company_change_clean"),
    ("mart.fact_company_change",
     "SELECT COUNT(*) FROM mart.fact_company_change"),
]
for label, sql in checks:
    n = con.execute(sql).fetchone()[0]
    print(f"{label}: {n:,} 筆")

# 事實表是否每家公司都至少有 1 筆變更（僅檢查最近 30 天內）

n_no_change = con.execute("""
    SELECT COUNT(*) FROM mart.dim_company d
    WHERE NOT EXISTS (
        SELECT 1 FROM mart.fact_company_change f
        WHERE f.uniform_no = d.uniform_no
        AND f.change_date >= current_date - INTERVAL '30 days'
    )
""").fetchone()[0]
print(f"30 天內沒有任何變更的公司數：{n_no_change:,}")
con.close()
# 輸出（依當日下載而略有不同）：

# raw.company_basic: 712,345 筆

# staging.company_basic_clean: 712,345 筆

# mart.dim_company: 712,345 家公司

# raw.company_change: 89,102 筆

# staging.company_change_clean: 89,102 筆

# mart.fact_company_change: 89,102 筆變更

# 30 天內沒有任何變更的公司數：687,234 家

這支驗證腳本做了兩件事：第一，比較三層的筆數是否一致（raw ≈ staging，staging 經 distinct 後 mart.dim_company 會比 staging 略少）；第二，統計「最近 30 天內沒有任何變更的公司家數」，這個指標可以用來監測「是不是資料沒進來」（如果整個市場的變更活動突然歸零，多半是平台當天出問題）。

如果想用一個 CTE 把兩張表 join 起來展示「公司基本資料 + 最近一次變更」：

"""de-journey/pipelines/join_demo.py：維度表 join 事實表的最小展示。"""
import duckdb

from pipelines.common import DUCKDB_PATH

con = duckdb.connect(str(DUCKDB_PATH))
result = con.execute("""
    WITH latest_change AS (
        SELECT uniform_no, change_date, change_item, after_value,
               ROW_NUMBER() OVER (
                   PARTITION BY uniform_no
                   ORDER BY change_date DESC
               ) AS rn
        FROM mart.fact_company_change
    )
    SELECT
        d.uniform_no,
        d.company_name,
        d.status,
        d.capital_amount,
        l.change_date AS latest_change_date,
        l.change_item AS latest_change_item
    FROM mart.dim_company d
    LEFT JOIN latest_change l
        ON d.uniform_no = l.uniform_no AND l.rn = 1
    WHERE d.capital_amount >= 100000000
    ORDER BY d.capital_amount DESC
    LIMIT 10
""").df()
print(result)
con.close()
# 輸出（依當日下載而略有不同）：

#   uniform_no  company_name  status  capital_amount  latest_change_date  latest_change_item

# 0  11000000   股份有限      active   5000000000      2024-11-20          資本額變更

# 1  12000000   股份有限      active   3000000000      2024-10-15          代表人變更

這個 CTE 範例展示了兩件事：第一，用 ROW_NUMBER() OVER (PARTITION BY uniform_no ORDER BY change_date DESC) 取出每家公司最近一次變更，搭配 WHERE rn = 1 過濾出唯一一筆；第二，LEFT JOIN 用在「沒有變更的公司也要列出來」的場景——這種公司通常是 latest_change_date 為 NULL 的「長期無變更」公司，分析時通常會另外標記。

常見錯誤與踩雷

錯誤一：QUALIFY 不是所有 SQL 引擎都支援。常見症狀：把 DuckDB 的 SQL 抄到 PostgreSQL 18 或 BigQuery 上跑，QUALIFY 直接報語法錯誤。對應排查方向：QUALIFY 是 SQL 2003 標準但實作不普遍，要改用 WITH ranked AS (... ROW_NUMBER() OVER ...) SELECT * FROM ranked WHERE rn = 1 的 CTE 寫法。DuckDB 1.4 完整支援 QUALIFY，所以本系列範例都能跑；跨平台時要記得改寫。

錯誤二：TRY_CAST 與 CAST 混用導致部分轉換失敗。常見症狀：資本額欄位有些是數字、有些是「未揭露」字串，CAST 會讓整個 SQL 失敗。對應排查方向：把所有對外來字串欄位做型別轉換的地方都改用 TRY_CAST，並接受「無法轉換的值變成 NULL」這個語意。

錯誤三：民國年公式在跨年時出錯。常見症狀：75 年 5 月 1 日（西元 1986）算成 1986-05-01 是對的，但「1051231」這種 7 位數（民國 105 年 12 月 31 日）會變成 105-12-31 = 西元 105-12-31（差 1900 多年）。對應排查方向：用正規表示式先驗證「是否為 6 或 7 位數字」，再用 LEFT/CAST/SUBSTR 把年、月、日拆開轉換。上面範例已經用 ^[0-9]{{6,7}}$ 做了驗證；如果遇到 8 位數（含世紀的寫法），要再加分支援。

錯誤四：WHERE dt = ? 漏寫導致整張表被覆寫。常見症狀：原本 raw.company_basic 是當天快照，CREATE OR REPLACE TABLE ... AS SELECT ... FROM raw.company_basic 沒加 WHERE 條件，結果意外地把所有歷史資料都用最新一天蓋掉了。對應排查方向：raw 表本身是 CREATE OR REPLACE TABLE 寫入的當天快照，所以 staging 從 raw 取資料時不需要 WHERE 條件——但如果是從 Parquet 分區檔讀取（例如跨日重跑），就要明確加上 WHERE dt = ?。

錯誤五：維度表忘了 DISTINCT 導致重複。常見症狀：同一家公司在 staging.company_basic_clean 裡出現多次（例如不同月份的下載合併），結果 mart.dim_company 有重複的 uniform_no。對應排查方向：用 QUALIFY ROW_NUMBER() = 1 而不是省略 distinct；並在 Day 32 品質檢查時加上「維度表 uniform_no 必須唯一」的規則。

效能與實務提醒

轉換與建模的效能瓶頸通常在「join 與彙總」，而非「讀檔」。以經濟部公司登記資料為例，70 萬筆基本 × 9 萬筆變更的 join 在 DuckDB 上不到 1 秒，因為 DuckDB 的欄式引擎對這種「小維度表 join 大事實表」的場景特別拿手。但如果你把 staging.company_change_clean 擴大到 1,000 萬筆，ROW_NUMBER() OVER (PARTITION BY ...) 會需要排序整個 partition，時間會拉到 5–10 秒。這時候可以考慮加上 PARTITION BY uniform_no 的索引（用 DuckDB 的 CREATE INDEX 實驗功能）或預先 group by。

實務上的另一個取捨是「staging 要保留多久」。政府開放資料的 staging 表本質上是 raw 的清洗版，所以保留 7–14 天就夠了，超過的部分可以從 raw 重新生成。但 mart 表不一樣——它是分析用的最終結果，需要長期保留。我們建議的策略是：raw 留 30 天、staging 留 14 天、mart 留長期（至少 1 年）。

另一個工程建議：把 transform_basic.py、transform_change.py、build_marts.py 三支腳本串成一支 run_transform.sh 或 run_transform.py，確保它們按順序執行：先 basic 再 change，最後 build marts。並且每次跑前都用 DROP TABLE IF EXISTS staging.X 確保冪等性（雖然 CREATE OR REPLACE 已經夠冪等，但對開發階段來說 DROP + CREATE 比較好除錯）。

小結

今天把昨天的落地資料進一步整理成「分析友善」的星狀結構。我們擴充了 pipelines/common.py，加入 staging_table、dim_table、fact_table 與 QUALITY_RULES；用 transform_basic.py 與 transform_change.py 把 raw 轉成 staging；用 build_marts.py 把 staging 組合成維度表 mart.dim_company 與事實表 mart.fact_company_change；最後用 verify_marts.py 驗證三層的一致性。重點回顧：第一，分層命名（raw → staging → mart）讓 SQL 可獨立測試與重跑；第二，TRY_CAST 與 QUALIFY 是 DuckDB 1.4 的兩個關鍵語法，分別處理缺失值與重複資料；第三，民國年轉西元年的公式要記得驗證 6/7 位數的格式；第四，維度表用 PARTITION BY uniform_no 確保唯一；第五，事實表的 surrogate key 用 ROW_NUMBER() 產生，後續儀表板查詢會更方便。

明天 Day 32 會做「品質檢查與監控」：用今天建立的 QUALITY_RULES 跑規則檢查、把結果寫到 meta.quality_check_result 表、並設定基本門檻（筆數不可低於昨天的 90%）。這是端到端管線「不能跑就好、要跑得對」的核心環節。

結語

今天的重點是「把 raw 升級成 staging 與 mart」。我們用 SQL 完成大部分工作，避免在 Python 層寫複雜的資料處理邏輯——這是 DuckDB 與現代分析工程的核心理念：能用 SQL 做的事不要拉進 Python。轉換層的程式碼量比昨天少很多，但「資料語意」的提升幅度比昨天大得多，因為維度表與事實表讓後續的分析查詢變得簡單直觀。

明天，我們會在今天的 mart 之上加一層品質檢查：對每一條 QUALITY_RULES 跑 SQL 統計違規筆數、把結果寫到 metadata 表、並在違規時主動觸發警報。這是 Day 33「失敗通知」的基礎，也是讓管線「從能跑變得可靠」的一步。我們會用一個可以整段執行的範例，從「全部通過」到「故意塞入髒資料看品質檢查是否能抓到」的兩個情境都展示一次。

延伸資源

  • DuckDB SQL 官方文件（2025，1.4 版）：https://duckdb.org/docs/stable/sql/introduction。本篇用到的 TRY_CAST、QUALIFY、ROW_NUMBER() OVER (...) 與 DATE 型別運算，皆以此文件為準。
  • Kimball 維度建模官方網站（2025）：https://www.kimballgroup.com/data-warehouse-business-intelligence-resources/kimball-techniques/dimensional-modeling-techniques/。本篇的 dim_company 與 fact_company_change 是典型的 Type 1 SCD + 事實表設計。
  • dbt 官方文件（2025，1.10 版）：https://docs.getdbt.com/。dbt 的 source、staging、marts 分層命名與本篇的 raw/staging/mart 概念一致，可以在本系列的 raw → staging → mart 之上加一層 dbt 模型。
  • DuckDB TRY_CAST 行為說明（2025）：https://duckdb.org/docs/stable/sql/expressions/cast。TRY_CAST 在轉換失敗時回傳 NULL，不會 raise，這是處理開放資料髒字串的標準做法。
  • 政府資料開放平臺公司登記資料檢視頁：https://data.gov.tw/。本系列使用的「公司登記資料」與「公司變更登記資料」位於此平台，授權為「政府資料開放授權條款第 1 版」。
  • Day 23 維度建模與 Day 24 維度建模實戰：分別介紹 star schema 的概念與實作。本篇是這兩篇的延伸，把抽象概念落實到真實管線。
標籤：

留言

張貼留言

這個網誌中的熱門文章

Day 2 變數與資料型別

Day 2 變數與資料型別 引言 寫程式的過程中，變數與資料型別是處理資料的基礎。變數是存放資料的容器，資料型別則決定這筆資料有哪些特性、可以進行哪些操作。學會定義變數、認識各種資料型別，是學好 Python 的關鍵一步。 這篇文章會帶你了解 Python 中變數的觀念、如何定義變數，以及常見的資料型別，包括整數、浮點數、字串、布林值，還有串列、元組、字典與集合等容器型別。我們也會介紹變數的命名規則與撰寫風格建議，以及如何用 type() 檢查資料型別。 什麼是變數？如何在 Python 中定義變數 變數是在程式執行時用來存放資料的名稱。透過定義變數，我們可以給一筆資料一個名字，並在程式的其他地方用這個名字取用該筆資料。在 Python 中，變數不需要事先宣告型別，因為 Python 是動態型別語言，變數的型別由指定給它的值決定。 定義變數的基本語法 在 Python 中定義變數非常簡單，只要用賦值符號 = 把值指定給變數即可。例如： x = 5 # 定義變數 x，並把整數 5 賦值給它 name = "Alice" # 定義變數 name，並把字串 "Alice" 賦值給它 在這裡，x 是一個變數，被賦予整數 5；name 是另一個變數，被賦予字串 "Alice"。 變數的更新與覆寫 變數的值可以修改，也就是說，我們可以在程式的不同地方給同一個變數新的值。例如： x = 10 # x 最初被賦予 10 x = 15 # x 的值現在被更新為 15 這樣就能依照需求，在程式執行過程中靈活調整變數的值。 Python 的動態型別系統 Python 和某些靜態型別語言不同，定義變數時不需要宣告型別。賦值時，Python 會根據值自動判斷變數的型別。例如： x = 5 # x 是整數 x = 3.14 # x 變成浮點數 x = "Hi" # x 變成字串 同一個變數在程式執行過程中可以存放不同型別的值，這是 Python 的彈性之一。 常見資料型別 在 Python 中，資料型別決定我們可以對變數進行哪些操作...
張貼留言
閱讀完整內容

Day 1 Python 簡介與環境設定

Day 1 Python 簡介與環境設定 引言 在現在的科技環境裡，程式設計已經是一項重要技能。無論你是對資料科學有興趣、想成為開發者，或是想踏入人工智慧（AI）領域，學會寫程式都能明顯提升你的競爭力。在眾多程式語言中，Python 因為語法簡單、功能強大、應用範圍廣泛，成為許多人進入程式世界的第一選擇。這篇文章會帶你認識 Python 的背景與優勢，並一步步教你在不同系統上安裝與設定 Python 開發環境，最後寫出第一支 Python 程式。 為什麼選擇 Python？ Python 是一種高階程式語言，由 Guido van Rossum 在 1991 年發布。Python 的設計哲學強調程式碼的可讀性，並用縮排來定義程式區塊，這點和許多使用大括號的語言不同。簡潔的語法讓它成為初學者的理想選擇；就算是經驗豐富的開發者，也能用它完成複雜的專案。 Python 的優勢如下： 簡單易學 ：Python 的語法清楚、結構簡潔，初學者很快就能上手。和其他語言相比，學習曲線相對平緩，不需要先弄懂一堆複雜觀念，就能開始寫程式。 應用範圍廣泛 ：從資料科學、網頁開發、人工智慧、機器學習、自動化測試到網路爬蟲，Python 都有大量開源函式庫與工具支援，而且在這些領域都扮演關鍵角色。 豐富的函式庫與框架 ：Python 的函式庫生態系非常龐大。做資料分析有 NumPy、Pandas；開發網站有 Django、Flask；做深度學習有 TensorFlow、PyTorch。各種需求幾乎都能找到對應的套件，讓開發更有效率。 跨平台支援 ：Python 支援 Windows、macOS、Linux 等作業系統，程式通常不需要太多修改就能跨平台執行，讓開發與部署更有彈性。 活躍的社群 ：Python 擁有龐大的開發者社群。學習或開發上遇到問題，幾乎都能在社群與論壇（例如 Stack Overflow）找到答案，對初學者來說是很強的後盾，也能減少卡關時的挫折感。 Python 的應用領域 Python 的流行與強大功能，讓許多領域都開始大量使用它。以下是幾個常見的應用方向： 資料科學 ：隨著大數據與人工智慧興起，資料科學大量使用 Python。NumPy、Pandas 與 Matplotlib 等工具能處理和分析龐...
張貼留言
閱讀完整內容

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門 隨著人工智慧（AI）與深度學習（Deep Learning）快速發展，越來越多人對這些技術產生興趣。不論你是想踏入 AI 領域的初學者，還是已經有程式基礎的開發者，學好 Python 與深度學習框架（例如 PyTorch），都能為你打開更多可能。 為什麼選擇 Python？ Python 已經是資料科學與人工智慧領域的首選語言。它的語法簡潔、容易上手，而且擁有龐大的生態系與大量開源函式庫。無論是資料處理、資料視覺化，還是建立機器學習與深度學習模型，Python 都能勝任。對想進入 AI 或資料科學領域的人來說，它幾乎是必備工具。 PyTorch 是什麼？ PyTorch 是由 Meta（原 Facebook）AI 研究團隊開發的開源深度學習框架，以易用、靈活和動態計算圖著稱，是許多 AI 研究人員與開發者的首選。相較於其他框架，PyTorch 的寫法更貼近原生 Python，對初學者相對友善。無論是簡單的實驗，還是複雜的深度學習模型，PyTorch 都能提供強大的支援。 這個系列能帶給你什麼？ 這個系列會從 Python 的基礎開始，帶你一步一步學習，最後能自己用 PyTorch 建立深度學習模型。即使你完全沒有寫過程式，也能跟著文章的節奏累積技能，理解 AI 與深度學習的核心觀念。 本系列涵蓋的主題 Python 基礎：從變數、條件判斷到函式與模組。 資料處理工具：用 NumPy 與 Pandas 有效率地操作資料。 資料視覺化：用 Matplotlib 與 Seaborn 把資料畫成圖表。 深度學習的數學基礎：線性代數、微積分與機率。 PyTorch 入門：理解張量、模型建構與 GPU 加速。 基礎深度學習模型：CNN 與 RNN 的實作應用。 深度學習專案實戰：從資料前處理到模型部署的端到端流程。 誰適合這個系列？ 程式初學者 ：如果你對 AI 充滿好奇，卻還沒寫過程式，系列的第一部分會帶你快速上手 Python，並幫助你理解深度學習的基本觀念。 資料科學愛好者 ：如果你已經熟悉一些資料處理方法，進階部分會教你如何用 PyTorch 建構深度學習模型。 開發者與研究人員 ：想更深入了...
張貼留言
閱讀完整內容