可靠資料產品：驗證、部署、觀測與復原

45 分鐘

Scheduling、Idempotency 與 Incremental Runs

學習目標：設計安全重跑；理解 full refresh 與 incremental processing。

排程在凌晨 2 點跑你的月度 pipeline，中途因為網路問題失敗了一半。值班的人手動重新觸發了一次，結果早上你發現輸出表裡同一個 batch_id 被算了兩次，總額整整多了一倍。這是典型的非 idempotent pipeline：重跑同一批輸入，卻沒有把之前寫入的結果覆寫或替換，而是疊加上去。設計安全的重跑，是任何排程化 pipeline 的基本要求，不是進階選配。

# 錯誤做法：每次執行都用 append，重跑會造成重複列
result.to_sql("monthly_summary", con=engine, if_exists="append")

# 正確做法：以 run 的處理範圍為 key，先刪除同一範圍的舊結果，再寫入新結果（idempotent）
def publish_idempotent(result: pd.DataFrame, period: str, engine) -> None:
    with engine.begin() as conn:
        conn.execute(
            "DELETE FROM monthly_summary WHERE period = :period",
            {"period": period}
        )
        result.to_sql("monthly_summary", con=conn, if_exists="append", index=False)

# incremental run 需要追蹤已處理到哪個時間點（watermark）
def get_watermark(engine) -> str:
    row = engine.execute("SELECT MAX(period) AS last_period FROM monthly_summary").first()
    return row.last_period if row and row.last_period else "1970-01"

1. 1確認寫入方式是否 idempotent
   同樣的輸入重跑兩次，輸出表的內容應該完全相同，不應該重複累加。
2. 2選擇 full refresh 或 incremental
   依資料量、來源是否會回溯修正，決定要整批重算還是只算新增部分。
3. 3追蹤 watermark
   incremental pipeline 需要記錄已處理到哪個時間點，作為下次執行的起點。
4. 4處理 late-arriving data
   設計機制讓比預期晚到的資料，仍能被正確涵蓋進對應期間的結果。

重點: 能完成步驟不代表結果可信；本節的能力必須能被驗證、重做並清楚說明。

45 分鐘

Logging、Metrics、Lineage 與 Observability

學習目標：記錄 run ID、row counts、freshness、validation、failure、lineage。

你的 pipeline 顯示昨晚的排程「成功」，但主管抱怨月報表的數字比平常少了九成。你打開 log，只看到一行「Job completed successfully」，沒有任何列數、沒有任何驗證結果，你完全無從判斷問題出在哪一個 stage。如果只記錄「成功／失敗」這種二元狀態，你等於在沒有儀表板的情況下開車：車子還在動，但你不知道油還剩多少、引擎溫度是否正常。

import logging, uuid

logger = logging.getLogger("ds301.pipeline")

def run_pipeline(input_path: str) -> None:
    run_id = str(uuid.uuid4())
    logger.info("run_start", extra={"run_id": run_id})

    raw = load_raw(input_path)
    logger.info("ingest_done", extra={"run_id": run_id, "row_count": len(raw)})

    errors = validate_contract(raw)
    logger.info("validation_done", extra={"run_id": run_id, "error_count": len(errors)})

    transformed = fill_missing_values(filter_current_schema(raw, "v2"))
    summary = summarize_by_metric(transformed)
    logger.info(
        "analyze_done",
        extra={"run_id": run_id, "output_row_count": len(summary), "total_value": float(summary["value"].sum())}
    )

    logger.info("run_end", extra={"run_id": run_id, "status": "succeeded"})

• row count 相較上次執行的變化超過一個合理門檻（例如 ±30%）
• validation error 數量從 0 變成非 0，即使目前只是 warning 等級
• 同一個 run_id 的 freshness 落後排程期望的時間超過一定門檻
• lineage 記錄顯示這次用了預期之外的輸入版本或程式碼版本

重點: 能完成步驟不代表結果可信；本節的能力必須能被驗證、重做並清楚說明。

45 分鐘

Human Approval、Publishing 與 Rollback

學習目標：高影響發布前設 approval，保留前版產物並定義 rollback。

你的 pipeline 全自動排程、全自動發布到公司內部儀表板，上個月因為上游一個欄位定義悄悄改變，發布出去的月度指標整整少算了一個事業群，而且是全自動流程，沒有任何人在發布前看過一眼。等到主管在會議上發現數字不對，錯誤的報表已經流通了兩週。高影響的資料產品，需要一個人在「發布給使用者」這一步之前確認結果合理，而不是完全信任自動化。

def publish_with_approval(result: pd.DataFrame, period: str, approved_by: str | None) -> None:
    if approved_by is None:
        # 高影響輸出：沒有人核准就不能發布，先寫入待審核區
        write_to_pending_review(result, period)
        raise RuntimeError(f"period={period} 尚未核准，已寫入待審核區，等待人工確認")

    # 版本化輸出：保留可識別的前一版，才有辦法 rollback
    archive_previous_version(period)
    publish_idempotent(result, period, engine)
    log_publish_event(period, approved_by)

def rollback(period: str) -> None:
    restore_previous_version(period)
    log_rollback_event(period)

• 發布後發現上游資料在發布前已經悄悄變質，但沒有被 contract 檢查攔下
• 核准者事後反悔，認為這次的結果不應該對外發布
• 下游使用者回報數字與他們自己的紀錄明顯不一致
• 發現這次發布使用了錯誤的程式碼版本或設定

重點: 能完成步驟不代表結果可信；本節的能力必須能被驗證、重做並清楚說明。

45 分鐘

Testing、Failure Recovery 與 Runbook

學習目標：使用 unit/data/integration smoke tests；定義 retry、人工介入與 runbook。

凌晨三點，pipeline 因為上游 API 暫時逾時而失敗。排程設定了「失敗就無限 retry」，結果整晚不斷重試、不斷失敗，隔天早上你發現系統資源被耗盡，而真正的問題（上游 API 早就恢復了，但你的程式碼有另一個永久性的 bug）從來沒有被解決。可靠的工作流需要三件事同時存在：測試能在問題發生前攔下已知錯誤、有分寸的 retry 策略能區分暫時性與永久性失敗、以及一份 runbook 讓值班的人知道遇到失敗該怎麼辦。

import time

class TransientError(Exception): ...
class PermanentError(Exception): ...

def call_upstream_with_retry(fetch_fn, max_attempts: int = 3) -> dict:
    for attempt in range(1, max_attempts + 1):
        try:
            return fetch_fn()
        except TransientError as e:
            if attempt == max_attempts:
                raise
            time.sleep(2 ** attempt)  # 指數退避，避免持續打爆上游
        except PermanentError:
            # 永久性錯誤（例如資料格式壞掉）：retry 不會成功，直接記錄並轉交人工
            log_for_manual_intervention(fetch_fn.__name__)
            raise

1. 1區分錯誤類型
   把錯誤分成暫時性（可以 retry）與永久性（retry 沒有意義）。
2. 2設定 retry 上限
   避免無限 retry 消耗資源或卡住整個排程。
3. 3定義人工介入時機
   永久性錯誤或超過 retry 上限時，應該通知人而不是繼續嘗試。
4. 4撰寫 runbook
   寫下常見失敗情境、判斷方式、處置步驟與升級對象。

重點: 能完成步驟不代表結果可信；本節的能力必須能被驗證、重做並清楚說明。

60 分鐘

Independent Lab：可靠 Analytics Workflow

學習目標：獨立交付 prototype、architecture canvas、tests、observability、runbook 與 ownership model。

• 前往 DS 301 實作頁面
• 預覽下載實作版分析資料包（含挑戰，合成資料）
• 下載 Independent Lab 起始 Notebook / 範本
• 預覽下載資料字典（繁體中文）
• 預覽下載資料字典（英文）

1. 1Input Contract
   寫清楚這個 workflow 的輸入資料契約與允許的假設。
2. 2Executable Prototype
   交付一份可執行的 pipeline，涵蓋 ingest→validate→transform→analyze→publish 五個 stage。
3. 3Tests
   至少包含一個 unit test 與一個 data validation 檢查。
4. 4Observability
   記錄 run ID、row counts、validation 結果與 lineage。
5. 5Human Approval & Rollback
   說明發布前的人工核准點，以及如何回退到前一版輸出。
6. 6Runbook
   寫下至少兩種常見失敗情境的判斷方式、處置步驟與升級對象。
7. 7Ownership Model
   明確寫出誰是這個 data product 的 owner，以及交接時需要知道的事。

• 只做了 happy path 測試，沒有測試資料違反 contract 時 pipeline 是否真的會失敗
• observability 只記錄成功/失敗，沒有記錄 row count 或 validation 結果
• runbook 寫得很籠統（例如「聯繫工程師」），沒有具體的判斷步驟與升級對象
• ownership model 只寫了名字，沒有說明交接時需要知道哪些背景知識

重點: 能完成步驟不代表結果可信；本節的能力必須能被驗證、重做並清楚說明。