跳到主要內容

DE Day 32 端到端管線（三）：品質檢查與監控

DE Day 32 端到端管線（三）：品質檢查與監控

執行需求：CPU 可跑。今天是端到端管線系列的第三天。昨天我們把 raw 整理成 staging 與 mart，組成了維度表 mart.dim_company 與事實表 mart.fact_company_change。今天要為這條管線加上「品質檢查」與「監控」：用 Day 31 擴充的 QUALITY_RULES 跑 SQL 規則檢查、把違規筆數寫到 meta.quality_check_result 表、並設定基本門檻（筆數不可低於昨天的 90%）。當品質異常時會留下紀錄，Day 33 會進一步把這份紀錄變成「失敗通知」。本篇所有範例都在 CPU 上執行。讀完這篇，你會有一套「自動檢查管線健康度」的腳本，每天早上跑一次就能知道昨天有沒有出問題。

引言

管線跑起來之後，「能跑」只是第一步，「跑得對」才是真正的考驗。我們在 Day 14 介紹過品質檢查的觀念：必填欄位不能 NULL、唯一鍵不能重複、數值必須在合理區間、日期必須能解析。今天要把那些觀念實作成「管線內建的一個步驟」，讓它每天自動跑，而不是等到下游使用者回報「報表怪怪的」才發現。

品質檢查的核心設計是「規則 + 違規筆數 + 門檻」。每條規則是一段 SQL 條件，例如 uniform_no IS NOT NULL；當這條 SQL 的違規筆數超過預設門檻（例如 0 或 100），就記錄到 meta.quality_check_result。這個設計的好處是「規則可累積、違規可追蹤、門檻可調整」。我們今天會用 Day 31 擴充的 QUALITY_RULES 字典當來源，把規則套用在 staging 與 mart 上。

監控則是「品質檢查的時間序列視角」。當我們每天跑品質檢查、把結果寫到 meta 表，幾天後就能看到「違規筆數的趨勢」。例如某一天 company_basic 的 uniform_no_length_8 違規筆數突然從 0 變 5 萬，那很可能是政府平台改了統一編號格式——這就是「資料變化的早期訊號」。我們會用一個簡單的 SQL 範例展示如何拉出最近 7 天的品質趨勢，並提醒讀者哪些指標值得日常關注。

品質檢查與監控的核心觀念

品質檢查通常分為四個層次，由淺到深依序是「schema 檢查」、「值域檢查」、「業務規則檢查」、「跨表一致性檢查」。schema 檢查是最基本的：欄位是否存在、型別對不對、NULL 比例有沒有異常。值域檢查是欄位值的範圍，例如統一編號必須是 8 碼、資本額必須是非負整數、日期必須在西元 1900 年之後。業務規則檢查是特定領域的規則，例如「公司名稱長度至少 2 個字」、「變更日期不能晚於今天」。跨表一致性檢查是表與表之間的關聯，例如「事實表 uniform_no 都必須在維度表中存在」、「每天新增的筆數不可低於昨天的 90%」。我們今天會覆蓋前兩層與第三層，第四層留給 Day 33 的「跨表檢查」與 Day 38 的「資料合約」。

另一個重要觀念是「品質門檻分級」。不是所有違規都同等嚴重——uniform_no IS NULL 是嚴重違規（會直接破壞 join），而 company_name 長度 >= 2 是輕微違規（可能是罕見的單字公司名）。我們會用 severity 欄位把違規分成 critical 與 warning 兩級，前者會觸發 Day 33 的通知、後者只留紀錄不通知。這個分級讓「雜訊」與「訊號」分開，避免被輕微違規淹沒真正的故障。

監控的另一個設計是「基準線（baseline）」。如果沒有過去的資料，當我們看到「違規筆數 100」時無法判斷這算多還算少。我們會建議保留最近 30 天的品質檢查結果，每天對照昨天、前天、同一週的上週同期，這樣「違規筆數突然變多」就能立刻被注意到。DuckDB 的視窗函式與 lag/lead 在這裡很好用，會在今天的範例中展示。

共用設定：擴充 common.py 加上門檻與嚴重度

先把 QUALITY_RULES 擴充，每條規則加上 severity（critical 或 warning）與 threshold（違規筆數門檻）：

"""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",
        "staging_table": "staging.company_basic_clean",
        "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",
    },
}

# 品質規則：每條 = (規則名稱, SQL 條件, 嚴重度, 門檻)

# 嚴重度 critical 會觸發 Day 33 通知，warning 只留紀錄

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

# 每日筆數門檻：今天的筆數不可低於昨天的 90%

MIN_DAILY_ROW_RATIO = 0.90

HTTP_TIMEOUT_SEC = 30
RETRY_ATTEMPTS = 3
RETRY_BACKOFF_SEC = 2.0

這份擴充有三個重點：第一，每條規則從 2 個欄位（名稱、SQL 條件）擴充到 4 個欄位（名稱、SQL 條件、嚴重度、門檻）；第二，新增 MIN_DAILY_ROW_RATIO = 0.90 當「日對日筆數」門檻，這是跨日監控的標準做法；第三，保留 Day 30–Day 31 的所有設定，確保管線向下相容。

完整實作：品質檢查腳本

接下來寫 pipelines/check_quality.py。這支腳本對每個資料集的每條規則跑違規筆數統計、把結果寫到 meta.quality_check_result、並在 stdout 印出違規摘要。執行前不需要安裝新套件（DuckDB 已內建）。

"""de-journey/pipelines/check_quality.py：對 staging 表跑品質規則並記錄結果。"""
import logging
import sys
from datetime import date
from pathlib import Path

import duckdb

from pipelines.common import (
    DATASETS,
    DUCKDB_PATH,
    LOGS_DIR,
    QUALITY_RULES,
)

LOGS_DIR.mkdir(parents=True, exist_ok=True)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    handlers=[
        logging.FileHandler(LOGS_DIR / f"quality-{date.today():%Y-%m-%d}.log"),
        logging.StreamHandler(sys.stdout),
    ],
)
log = logging.getLogger("quality")


def ensure_meta_tables(con: duckdb.DuckDBPyConnection) -> None:
    """建立品質檢查結果表，idempotent。"""
    con.execute("CREATE SCHEMA IF NOT EXISTS meta")
    con.execute("""
        CREATE TABLE IF NOT EXISTS meta.quality_check_result (
            check_date DATE,
            dataset VARCHAR,
            rule_name VARCHAR,
            severity VARCHAR,
            threshold BIGINT,
            violation_count BIGINT,
            is_breached BOOLEAN,
            PRIMARY KEY (check_date, dataset, rule_name)
        )
    """)


def run_rule(con: duckdb.DuckDBPyConnection, dataset: str, rule_name: str,
             sql_cond: str, severity: str, threshold: int) -> int:
    """對指定資料集跑單一規則，回傳違規筆數。

    sql_cond 是「合規的條件」；違規 = NOT(合規)。
    """
    cfg = DATASETS[dataset]
    target_table = cfg["staging_table"]
    violation_sql = f"""
        SELECT COUNT(*) FROM {target_table}
        WHERE NOT ({sql_cond})
    """
    n_violation = con.execute(violation_sql).fetchone()[0]
    is_breached = n_violation > threshold
    con.execute("""
        INSERT OR REPLACE INTO meta.quality_check_result
        VALUES (current_date, ?, ?, ?, ?, ?, ?)
    """, [dataset, rule_name, severity, threshold, n_violation, is_breached])
    return n_violation


def run_dataset(con: duckdb.DuckDBPyConnection, dataset: str) -> dict:
    """對單一資料集跑所有規則，回傳違規摘要。"""
    rules = QUALITY_RULES[dataset]
    summary = {"critical_breaches": 0, "warning_breaches": 0, "details": []}
    for rule_name, sql_cond, severity, threshold in rules:
        n = run_rule(con, dataset, rule_name, sql_cond, severity, threshold)
        breached = n > threshold
        if breached:
            if severity == "critical":
                summary["critical_breaches"] += 1
            else:
                summary["warning_breaches"] += 1
        summary["details"].append(
            f"  [{severity}] {rule_name}: {n} 筆違規（門檻 {threshold}）"
            + (" *** 超標 ***" if breached else "")
        )
    return summary


def run_row_count_check(con: duckdb.DuckDBPyConnection, dataset: str) -> bool:
    """跨日筆數檢查：今天的筆數不可低於昨天的 90%。"""
    from pipelines.common import MIN_DAILY_ROW_RATIO
    cfg = DATASETS[dataset]
    target = cfg["staging_table"]
    rows = con.execute(f"""
        WITH today_count AS (
            SELECT COUNT(*) AS n FROM {target}
            WHERE DATE_TRUNC('day', current_timestamp) = current_date
        ),
        yesterday_count AS (
            SELECT n FROM meta.daily_row_count
            WHERE dataset = ? AND check_date = current_date - INTERVAL '1 day'
        )
        SELECT
            (SELECT n FROM today_count) AS today_n,
            (SELECT n FROM yesterday_count) AS yesterday_n
    """, [dataset]).fetchone()
    today_n, yesterday_n = rows[0], rows[1]
    if yesterday_n is None or yesterday_n == 0:
        return False  # 無基準線，不檢查

    ratio = today_n / yesterday_n
    breached = ratio < MIN_DAILY_ROW_RATIO
    con.execute("""
        INSERT OR REPLACE INTO meta.quality_check_result
        VALUES (current_date, ?, 'daily_row_count', 'critical', ?, ?, ?)
    """, [dataset, int(yesterday_n * MIN_DAILY_ROW_RATIO), today_n, breached])
    return breached


def main() -> int:
    con = duckdb.connect(str(DUCKDB_PATH))
    try:
        ensure_meta_tables(con)
        any_critical = False
        for dataset in DATASETS:
            log.info("=== %s 品質檢查 ===", dataset)
            summary = run_dataset(con, dataset)
            for line in summary["details"]:
                log.info(line)
            if run_row_count_check(con, dataset):
                log.warning("[%s] 筆數低於昨日的 %.0f%%",
                            dataset, MIN_DAILY_ROW_RATIO * 100)
                any_critical = True
            if summary["critical_breaches"] > 0:
                any_critical = True
        con.close()
        return 1 if any_critical else 0
    except Exception:
        con.close()
        raise


if __name__ == "__main__":
    sys.exit(main())

這支腳本的核心邏輯是 run_rule()：對每條規則，把 SQL 條件當成「合規條件」，然後查詢「不符合這個條件的筆數」。例如 uniform_no IS NOT NULL 是合規條件，違規就是 uniform_no IS NULL 的筆數。is_breached = n_violation > threshold 決定是否觸發警報；INSERT OR REPLACE INTO meta.quality_check_result 把結果寫到 metadata 表，這樣每天的違規都會累積成時間序列。

run_row_count_check() 是另一個關鍵函式：跨日筆數檢查。它的 SQL 邏輯是「今天的筆數 vs 昨天的筆數」，比率低於 MIN_DAILY_ROW_RATIO（預設 0.90）就算違規。實務上這個檢查特別有效——政府平台如果當天出問題，筆數會立刻腰斬或歸零，比欄位值錯誤更容易被發現。

腳本最後回傳 0 或 1 當 exit code，這是 Unix 工具鏈的標準做法：0 代表成功、非 0 代表失敗。Day 33 的排程器會用這個 exit code 決定是否觸發通知。

完整實作：品質趨勢監控

跑完今天的品質檢查後，我們想看「最近 7 天的違規趨勢」。這個查詢用 DuckDB 的視窗函式把今天的違規筆數和昨天、前天比較：

"""de-journey/pipelines/quality_trend.py：最近 7 天品質違規趨勢。"""
import duckdb

from pipelines.common import DUCKDB_PATH

con = duckdb.connect(str(DUCKDB_PATH))
trend = con.execute("""
    WITH recent AS (
        SELECT
            check_date,
            dataset,
            rule_name,
            severity,
            violation_count,
            is_breached
        FROM meta.quality_check_result
        WHERE check_date >= current_date - INTERVAL '7 days'
    ),
    with_lag AS (
        SELECT
            *,
            LAG(violation_count) OVER (
                PARTITION BY dataset, rule_name
                ORDER BY check_date
            ) AS prev_violation_count
        FROM recent
    )
    SELECT
        check_date,
        dataset,
        rule_name,
        severity,
        violation_count,
        prev_violation_count,
        violation_count - COALESCE(prev_violation_count, 0) AS delta
    FROM with_lag
    WHERE is_breached OR (prev_violation_count IS NOT NULL AND violation_count > prev_violation_count * 2)
    ORDER BY check_date DESC, dataset, rule_name
""").df()
print(trend.to_string(index=False))
con.close()
# 輸出（依當日資料而略有不同）：

#   check_date  dataset        rule_name             severity  violation_count  prev_violation_count  delta

#  2025-12-06   company_basic  uniform_no_length_8   critical              12                   0    12

#  2025-12-06   company_basic  daily_row_count       critical          650000             712000 -62000

這段 SQL 用到了 Day 4 講過的 LAG() 視窗函式：對「同一個資料集同一條規則」的時間序列，取前一天的違規筆數當基準。WHERE ... OR (prev_violation_count IS NOT NULL AND violation_count > prev_violation_count * 2) 這段是「異常偵測」：除了「已被標記為 breached」的違規，也把「比前一天翻倍」的情況撈出來——這種「突然增加」往往是資料源在當天做了改動的早期訊號。

把這份趨勢表用 pandas 或 Streamlit 視覺化（Day 34 會接上）就能看到「品質儀表板」。實務上我們會建議把這份查詢的結果存成 meta.quality_trend_daily 當快取，避免每次開儀表板都重算。

常見錯誤與踩雷：補上更多檢查腳本

在介紹常見錯誤之前，先把幾支常用的「小型檢查腳本」補齊。這些腳本不寫入 metadata、只在 stdout 印出違規樣本，適合臨時排查問題時使用。

"""de-journey/pipelines/dump_violations.py：把違規樣本倒出來看。"""
import duckdb

from pipelines.common import DUCKDB_PATH

con = duckdb.connect(str(DUCKDB_PATH))
# 取出統一編號不是 8 碼的 5 筆樣本

rows = con.execute("""
    SELECT uniform_no, company_name, LENGTH(uniform_no) AS len
    FROM staging.company_basic_clean
    WHERE LENGTH(uniform_no) <> 8
    LIMIT 5
""").fetchall()
for r in rows:
    print(r)  # 輸出：tuple 形式的違規樣本

con.close()
# 輸出範例：

# ('1234567', '○○有限公司', 7)

# ('123456789', '△△有限公司', 9)

這支小腳本只做一件事：把違規的明細倒出來。比 metadata 表的「違規筆數」更有用的是「違規長什麼樣」，實務上 80% 的品質問題可以靠看 5 筆樣本判斷成因。

"""de-journey/pipelines/compare_yesterday.py：今天的資料 vs 昨天的資料差異。"""
import duckdb

from pipelines.common import DUCKDB_PATH

con = duckdb.connect(str(DUCKDB_PATH))
diff = con.execute("""
    WITH today AS (
        SELECT uniform_no, company_name, capital_amount,
               hash(concat(uniform_no, company_name, capital_amount::varchar)) AS h
        FROM mart.dim_company
        WHERE DATE_TRUNC('day', current_timestamp) = current_date
    ),
    yesterday AS (
        SELECT uniform_no, company_name, capital_amount,
               hash(concat(uniform_no, company_name, capital_amount::varchar)) AS h
        FROM mart.dim_company
        WHERE DATE_TRUNC('day', current_timestamp) = current_date - INTERVAL '1 day'
    )
    SELECT
        COALESCE(t.uniform_no, y.uniform_no) AS uniform_no,
        t.company_name AS today_name,
        y.company_name AS yesterday_name,
        t.capital_amount AS today_cap,
        y.capital_amount AS yesterday_cap,
        CASE
            WHEN t.h IS NULL THEN 'deleted'
            WHEN y.h IS NULL THEN 'inserted'
            WHEN t.h <> y.h THEN 'changed'
        END AS diff_type
    FROM today t
    FULL OUTER JOIN yesterday y USING (uniform_no)
    WHERE t.h IS DISTINCT FROM y.h
    ORDER BY uniform_no
    LIMIT 20
""").df()
print(diff)
con.close()
# 輸出（依當日資料而略有不同）：

#   uniform_no  today_name  yesterday_name  today_cap  yesterday_cap  diff_type

#  11000000      ...         ...             5000000000  4000000000     changed

這支腳本用 FULL OUTER JOIN 把「新增、刪除、修改」三種變化都抓出來，搭配 hash() 把整列壓成單一雜湊值，比較兩個日期快照的差異。這對「為什麼今天的維度表跟昨天不同？」這個常見問題非常有用。當 diff_type 全是 changed 但 company_name 完全相同時，就代表問題出在 capital_amount 上；當 diff_type 出現大量 deleted 時，就要懷疑資料源是否有大規模刪除。

"""de-journey/pipelines/quality_dashboard.py：把今天品質結果摘要印出。"""
import duckdb

from pipelines.common import DUCKDB_PATH

con = duckdb.connect(str(DUCKDB_PATH))
summary = con.execute("""
    SELECT
        dataset,
        SUM(CASE WHEN severity='critical' AND is_breached THEN 1 ELSE 0 END) AS n_critical_breached,
        SUM(CASE WHEN severity='warning' AND is_breached THEN 1 ELSE 0 END) AS n_warning_breached,
        SUM(CASE WHEN is_breached THEN violation_count ELSE 0 END) AS total_violations
    FROM meta.quality_check_result
    WHERE check_date = current_date
    GROUP BY dataset
    ORDER BY dataset
""").df()
print(summary)
con.close()
# 輸出範例：

#           dataset  n_critical_breached  n_warning_breached  total_violations

#  company_basic                   0                    1               1234

#  company_change                  0                    0                  0

這支腳本是「每日品質儀表板的純文字版」。當你在終端機裡排程跑完一輪檢查，最後呼叫這支腳本就能在 stdout 看到「今天哪個資料集、有幾條 critical 違規、有幾條 warning 違規」。Day 34 會把這個摘要用 Streamlit 視覺化。

常見錯誤與踩雷

錯誤一：忘記把 staging 表建好就跑品質檢查。常見症狀：Catalog Error: Table with name staging.company_basic_clean does not exist。對應排查方向：品質檢查的目標是 staging 與 mart，所以跑 check_quality.py 之前一定要先跑 Day 30 的 ingest.py 與 Day 31 的 transform_*.py 與 build_marts.py。在排程器裡，這三步要按順序串起來，Day 36 / Day 37 會展示排程的具體寫法。

錯誤二：門檻設成 0 結果任何違規都觸發警報。常見症狀：每天早上信箱被幾十封「critical」警報淹沒。對應排查方向：門檻要根據「合理範圍」設定。例如 establish_date_valid 設為 warning 嚴重度、門檻 1000，因為政府平台偶爾會有幾筆「公司設立日期不公開」的個案，這種輕微違規不該觸發通知。對應 uniform_no_length_8 這種關鍵規則，才用 critical 嚴重度與 0 門檻。

錯誤三：meta.quality_check_result 的 PRIMARY KEY 重複導致 INSERT 失敗。常見症狀：Constraint Error: Duplicate key in primary key。對應排查方向：PRIMARY KEY 是 (check_date, dataset, rule_name)，所以同一天對同一個資料集的同一條規則只能有一筆。我們用 INSERT OR REPLACE 而非 INSERT 來避免這個問題；如果你改成普通 INSERT，記得先 DELETE 當天的紀錄再寫入。

錯誤四：LAG() 在 DuckDB 裡回傳 BIGINT，要小心型別轉換。常見症狀：violation_count - prev_violation_count 算出負數時變成 -0。對應排查方向：用 COALESCE(prev_violation_count, 0) 把 NULL 補成 0，再用一般整數運算；輸出時再用 int() 或 abs() 包裝。

錯誤五：跨日筆數檢查的基準線被覆寫。常見症狀：今天跟昨天的基準線永遠是「同一天」。對應排查方向：meta.daily_row_count 必須是「每天寫入一次、保留 30 天以上」的歷史表。實務上我們會在 check_quality.py 結束前另外寫一行 INSERT INTO meta.daily_row_count SELECT current_date, dataset, COUNT(*) FROM staging.X，把當天的筆數存進去，避免被覆寫。

效能與實務提醒

品質檢查的效能瓶頸在「對每條規則跑一次 SELECT COUNT(*) FROM ... WHERE NOT (condition)」。以 staging.company_basic_clean 的 70 萬筆資料、6 條規則計算，每次跑大約 3–6 秒。加上 staging.company_change_clean 的 9 萬筆與 3 條規則，整個品質檢查應該在 10 秒以內完成。這對每天一次的管線來說綽綽有餘。

實務上的另一個取捨是「規則數量」。我們今天列了 9 條規則，這是「品質檢查的最小集合」。隨著管線演進，規則會越來越多；當規則數量到 50 條以上時，可以考慮：

  1. 把規則依嚴重度分檔，只在「快速檢查」跑 critical 規則。
  2. 用 UNION ALL 把同一個資料集的多條規則合併成一個查詢，減少 SELECT COUNT(*) 的次數。
  3. 把品質檢查結果先寫成 Parquet，再載入 meta 表，縮短寫入時間。

另一個工程建議：把 check_quality.py 與 quality_trend.py 排成「每天早上 9 點跑」。這個時段通常前一天晚上的管線已經收尾，且使用者還沒進辦公室看到報表。在 9:00 跑、9:30 寄通知（Day 33 會做），使用者進辦公室時就知道昨天有沒有出問題，比「半夜三點跑、失敗訊息在信箱裡躺 8 小時」要好得多。

小結

今天為端到端管線加上了品質檢查與監控。我們擴充了 pipelines/common.py 的 QUALITY_RULES，把每條規則加上嚴重度與門檻；用 check_quality.py 對每個資料集跑規則、把違規筆數寫到 meta.quality_check_result；用 quality_trend.py 用視窗函式撈出最近 7 天的違規趨勢與「突然增加」的異常。重點回顧：第一，品質規則分四層（schema、值域、業務、跨表），今天覆蓋前三層，跨表留給 Day 33 與 Day 38；第二，門檻要根據合理範圍設定，避免被雜訊淹沒；第三，INSERT OR REPLACE 與明確的 PRIMARY KEY 設計確保結果表冪等；第四，跨日筆數檢查用「今天的筆數 vs 昨天的 90%」是最有效的「資料源異常」早期訊號；第五，LAG() 視窗函式是品質趨勢監控的核心，能抓到「突然增加」這類肉眼不易察覺的異常。

明天 Day 33 會接著做「失敗通知與重跑」：用今天的 meta.quality_check_result 當觸發條件、把 critical 違規組合成通知訊息、用模擬的發信程式（標示為模擬）寄出，並提供「單一資料集重跑」的腳本讓發生錯誤時不必整條管線重來。

結語

今天的重點是「把品質檢查變成管線的內建步驟」。我們沒有引入新的重型工具，而是用 DuckDB 內建的 SQL 與 meta 表完成所有事。這個設計的好處是「簡單、可移植、容易除錯」：當明天有人問「昨天的 uniform_no_length_8 違規 12 筆怎麼來的」，你可以直接用 SQL 撈出那 12 筆樣本，肉眼檢查為什麼統一編號不是 8 碼。

明天，我們會把今天的 meta.quality_check_result 接到一個「模擬發信」程式：當 critical 違規發生時，把違規摘要寫成一份 HTML 報告並「模擬」寄出。同時我們會提供「單一資料集重跑」的腳本與 CLI 介面，讓發生錯誤時不必整條管線重來。這是讓管線「從能跑、到跑得對、再到出問題時知道怎麼救」的最後一步。

延伸資源

  • Great Expectations 官方文件（2025，1.x 版）：https://docs.greatexpectations.io/。本篇用「規則 + 違規筆數 + 門檻」的設計與 Great Expectations 的 Expectation Suite 概念互通；如果你的團隊需要更完整的資料品質框架，可以評估 GE。
  • DuckDB 視窗函式官方文件（2025，1.4 版）：https://duckdb.org/docs/stable/sql/window_functions。本篇的 LAG() 與 PARTITION BY 用法以此文件為準。
  • DuckDB metadata 設計模式（2025）：https://duckdb.org/docs/stable/sql/meta。information_schema.columns 與 information_schema.tables 提供 schema 層級的查詢，能在 Day 38 的「資料合約」章節派上用場。
  • 政府資料開放平臺公司登記資料檢視頁：https://data.gov.tw/。本系列使用的「公司登記資料」與「公司變更登記資料」位於此平台，授權為「政府資料開放授權條款第 1 版」。
  • Day 14 資料品質檢查：觀念章節，介紹規則設計與自動化的設計原則。本篇是 Day 14 的延伸，把觀念落實到 DuckDB SQL 與 Python 腳本。
  • Day 22 失敗處理：失敗分類（瞬時、編碼、邏輯）與重試、退避策略。本篇的 retry 設計沿用 Day 22 的精神。
標籤：

留言

張貼留言

這個網誌中的熱門文章

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 建構深度學習模型。 開發者與研究人員 ：想更深入了...
張貼留言
閱讀完整內容