從 Notebook 到可重複分析工作流

40 分鐘

什麼是 Data Product / Analytics Workflow

學習目標：定義使用者、輸入、輸出、freshness、品質期望與 owner。

你每個月手動跑一份 notebook，產出一張給主管看的月報表。上個月你請假，同事臨時接手，卻不知道這份 notebook 該用哪一份原始資料、輸出要放在哪裡、多久要更新一次，也不知道出錯了要找誰。這份分析從來沒有真正「產品化」：它只活在你的腦中與你的筆電裡。Data product 的第一步，不是寫程式，而是把這些隱性知識寫下來：誰會用這份輸出、他們期待多新的資料、什麼品質才算可信、誰對它的正確性負責。

import pandas as pd

# clean：月度分析資料包，一列代表某個來源系統在某個 batch 下、某個實體的一筆指標觀測
df = pd.read_csv("DS301_Analytics_Package_Clean.csv", parse_dates=["event_time", "ingested_at"])
df.shape
df.dtypes
df[["run_id", "source_system", "batch_id", "metric", "value", "schema_version"]].head()

# 先問：這份資料的 owner 是誰？如果 schema_version 變了，誰該被通知？
df["schema_version"].unique()

• 只有你知道要去哪裡抓原始資料，流程全部記在腦袋裡
• 沒有人知道這份輸出多久要更新一次，也沒有人檢查它是否真的如期更新
• 「這個數字對不對」只能靠原作者口頭保證，沒有任何驗證紀錄
• 換一台機器或換一個人執行，就會因為路徑或帳密問題直接失敗

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

45 分鐘

Notebook → Functions → Stages

學習目標：將探索程式拆成具明確 input/output、可重跑的 stages。

探索階段的 notebook 常常是一長串由上到下執行的 cell：讀資料、清理、篩選、彙總、畫圖全部混在一起，變數互相依賴，改了第 3 格就要重跑後面所有格子。這種寫法適合探索，但不適合變成可靠的工作流：沒有人能只重跑「清理」這一步而不影響其他部分，也沒辦法針對某一步單獨寫測試。把探索程式拆成 functions 與 stages，是把「一長串魔法」變成「一串可以個別檢查、個別重跑的積木」。

# 探索階段：一長串互相依賴的 cell（難以重跑、難以測試）
df = pd.read_csv("DS301_Analytics_Package_Clean.csv")
df = df[df["schema_version"] == "v2"]
df["value"] = df["value"].fillna(0)
summary = df.groupby(["source_system", "metric"])["value"].sum()

# 拆成 stage：每個 function 有清楚的 input 與 output，可以個別測試、個別重跑
def load_raw(path: str) -> pd.DataFrame:
    return pd.read_csv(path, parse_dates=["event_time", "ingested_at"])

def filter_current_schema(df: pd.DataFrame, version: str) -> pd.DataFrame:
    return df[df["schema_version"] == version].copy()

def fill_missing_values(df: pd.DataFrame) -> pd.DataFrame:
    df["value"] = df["value"].fillna(0)
    return df

def summarize_by_metric(df: pd.DataFrame) -> pd.DataFrame:
    return df.groupby(["source_system", "metric"], as_index=False)["value"].sum()

1. 1找出自然的斷點
   在探索 notebook 中找出讀取、清理、篩選、彙總等自然階段的邊界。
2. 2為每個階段寫一個 function
   明確定義輸入型態與輸出型態，避免依賴全域變數。
3. 3移除隱性依賴
   確認每個 function 不依賴前一個 cell 遺留在記憶體中的變數。
4. 4為關鍵 function 加上測試
   至少為一個容易出錯的 stage（如清理或聚合）寫一個簡單測試。

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

40 分鐘

Configuration、Secrets 與 Environments

學習目標：設定與程式分離；禁止 hardcode secrets；理解 dev/test/prod。

同事把資料庫密碼直接寫進 notebook 的第一格，方便自己開發時貼上就能跑。三個月後這份 notebook 被分享給另一個團隊做教學示範，密碼也跟著一起外流。更常見的問題是：同一段程式碼在你的筆電上跑得動，換一台機器、換一個資料庫就整段失敗，因為路徑、帳密、資料表名稱全部寫死在程式碼裡。把 configuration 與 secrets 從程式碼中抽離，是讓同一份程式碼可以安全地在 dev、test、prod 三種環境下運作的前提。

# 錯誤做法：帳密與環境設定直接寫死在程式碼裡
DB_PASSWORD = "sup3r-secret-2026"  # 絕對不要這樣做
INPUT_PATH = "/Users/alice/Desktop/data.csv"

# 正確做法：從環境變數 / secret 管理機制讀取，程式碼本身不含任何機密
import os

DB_PASSWORD = os.environ["DS301_DB_PASSWORD"]  # 由 secret 管理機制注入，不寫入版本控制
ENVIRONMENT = os.environ.get("DS301_ENV", "dev")  # dev / test / prod
INPUT_PATH = os.environ["DS301_INPUT_PATH"]  # 依環境不同指向不同資料來源

1. 1找出所有寫死的值
   檢查程式碼中的密碼、路徑、資料表名稱是否直接寫在程式裡。
2. 2搬進環境變數或 config 檔
   把這些值改成從環境變數或設定檔讀取，程式碼本身不再包含機密。
3. 3區分 dev/test/prod 設定
   確保同一份程式碼可以透過不同的環境設定，在不同環境下正確運作。
4. 4確認 secret 不會被提交
   檢查版本控制設定，確保 secret 檔案不會被意外提交進版本紀錄。

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

45 分鐘

Data Contracts 與 Pipeline Validation

學習目標：schema/quality 不符時提早失敗，區分 warning 與 blocking error。

上游系統在沒有通知任何人的情況下，把 metric 欄位的其中一個值從 "conversion_rate" 拼成了 "converion_rate"。你的 pipeline 沒有檢查這件事，直接把資料丟進彙總邏輯，結果那筆指標被靜靜地歸到「其他」類別，月報表少算了一整條產品線的轉換率，而且完全沒有跳出任何錯誤。Data contract 就是把「這份資料應該長什麼樣子」寫下來，並在資料違反契約時提早、大聲地失敗，而不是讓錯誤的資料悄悄流進下游。

ALLOWED_METRICS = {"conversion_rate", "active_users", "avg_response_ms"}
REQUIRED_COLUMNS = {"run_id", "source_system", "batch_id", "event_time", "ingested_at", "entity_id", "metric", "value", "schema_version"}

def validate_contract(df: pd.DataFrame) -> list[str]:
    errors = []
    missing = REQUIRED_COLUMNS - set(df.columns)
    if missing:
        errors.append(f"缺少必要欄位（blocking）：{sorted(missing)}")

    unknown_metrics = set(df["metric"].unique()) - ALLOWED_METRICS
    if unknown_metrics:
        errors.append(f"出現未知的 metric 值（blocking）：{sorted(unknown_metrics)}")

    null_ratio = df["value"].isna().mean()
    if null_ratio > 0.2:
        errors.append(f"value 缺值比例過高（warning，需人工確認）：{null_ratio:.1%}")

    return errors

errors = validate_contract(df)
blocking = [e for e in errors if "blocking" in e]
if blocking:
    raise ValueError(f"Pipeline 因 contract 違反而提早失敗：{blocking}")

• 必要欄位清單與各欄位的資料型態
• 允許的類別值清單（例如 metric 可以出現哪些值）
• 數值範圍與缺值比例的可接受上限
• 違反規則時該視為 warning 還是 blocking error
• schema 變更時的通知流程與版本號規則

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

70 分鐘

Guided Lab：把 Notebook 改成 Pipeline

學習目標：在教師引導下，建立 ingest→validate→transform→analyze→publish stages 與 checkpoints。

• 前往 DS 301 實作頁面
• 預覽下載乾淨版分析資料包（CSV，合成資料）
• 下載 Guided Lab 起始 Notebook（Notebook → Pipeline 改寫）
• 預覽下載資料字典（繁體中文）
• 預覽下載資料字典（英文）

1. 1Ingest
   把讀取原始資料的邏輯獨立成一個 stage，只負責讀取，不做任何清理或計算。
2. 2Validate
   在資料進入後續處理前，套用 data contract 檢查，違反 blocking 規則時提早失敗。
3. 3Transform
   把清理、篩選、型態轉換等邏輯獨立成 stage，維持每個 function 的明確 input/output。
4. 4Analyze
   把彙總與計算邏輯獨立成 stage，並在此加入一個 checkpoint（例如列數、加總值）確認結果合理。
5. 5Publish
   把輸出寫入最終位置的邏輯獨立成 stage，並記錄這次執行使用的 config 與環境。

• 把 notebook 的 cell 直接複製成 function，卻忘了移除對全域變數的依賴
• validate stage 只是形式上存在，沒有真正套用 contract 檢查
• 把 config 或 secret 直接寫在 pipeline 程式碼裡，而不是從環境變數讀取
• 完全沒有設 checkpoint，只看 pipeline 有沒有報錯，不看中間結果是否合理

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