DE Day 41 專案定義與資料模型

執行需求：CPU 可跑。今天是專案篇的第一篇。我們要把過去 40 天學過的所有工具鏈（uv、DuckDB、Polars、dbt、APScheduler、Streamlit、Great Expectations、GitHub Actions）組合成一條每天自動運行的真實管線。這個專案的目標是「每天從 data.gov.tw 抓取行政院環境部公開的空氣品質監測資料，整理成可分析的維度模型，並用 Streamlit 儀表板呈現最近 30 天的趨勢」。整個專案在 4 個工作天（Day 41-44）內完成，並在 Day 45 做系列總結。今天的焦點是「先把範圍、資料、合約、模型定義清楚」，讓後續三天的實作有穩固的藍圖。

引言

專案第一天最常見的錯誤是「先寫程式再想需求」。當你打開編輯器開始寫 ingest 腳本時，如果還沒釐清「要抓什麼、為什麼抓、怎麼算成功」，後面三天幾乎一定會重寫。專案管理的經驗法則是「需求定義佔 30% 時間、實作佔 50%、測試與調整佔 20%」，但很多新手把時間倒過來，結果是程式寫了一堆、需求還沒對齊，最後整批打掉重來。今天我們刻意把 80% 的時間花在「想清楚」、20% 寫第一支 ingest 與 staging 模型，這樣明天到後天的進度會順很多。

本專案選擇「行政院環境部空氣品質監測資料」有幾個理由。第一，它是真實的政府開放資料，位於 data.gov.tw 平台、採政府資料開放授權條款第 1 版，授權清楚、再散布沒問題。第二，它每天更新、每小時有量測值，適合做「每日增量」的管線設計。第三，它有多個測站、多種污染物（AQI、PM2.5、PM10、O3、CO、SO2、NO2），可以同時示範「事實表 + 多維度表」的維度建模。第四，環境部有公開的 API（位於 airtw.moenv.gov.tw）與 CSV 兩種取得方式，方便我們做「排程抓 API、落地成 CSV、寫進 DuckDB」的標準 ETL 流程。

今天的內容分四段。第一段定義專案範圍與利害關係人；第二段定義資料來源、合約與維度模型；第三段建立 dbt 專案結構並寫出 staging 模型；第四段定義可重複執行的專案啟動腳本。明天（Day 42）會把 ingest 與排程接起來，Day 43 會做品質檢查與評估，Day 44 會做部署與儀表板。今天先把「資料合約」與「維度模型」這兩個基礎蓋好，這兩件事是後續四天的共用設定，所有 Day 41-45 的程式碼都引用同一份設定。

專案範圍與利害關係人

先把專案範圍寫成一份「產品需求文件」，避免實作到一半才發現「啊，這個不在這次範圍」。我們用一頁 A4 的篇幅定義目標、使用者、承諾、不承諾，以及成功指標。這份文件之後會放進 docs/scope.md，成為接下來四天的共同語言。

目標是建立一條每日自動運行的管線，從 data.gov.tw 抓取行政院環境部公開的逐時空氣品質量測，整理成維度模型，最後用 Streamlit 儀表板呈現。使用者分兩種：第一種是「資料分析師」，他們需要 SQL 查詢乾淨的事實表與維度表（不需要碰 ETL 程式碼）；第二種是「一般使用者」，他們透過 Streamlit 看最近 30 天的趨勢、各測站排名、各縣市比較。

承諾的事項有四個：每日 03:00 (UTC+8) 完成前一天的資料更新；儀表板能正確呈現最近 30 天的 AQI 趨勢；至少有 5 個關鍵指標（全國日均 AQI、不良日比例、PM2.5 月均、各測站排名、縣市不良率）的歷史資料；當合約破壞時自動寫進告警檔。不承諾的事項也明寫：不做即時（每小時）抓取，只做每日聚合；不做預測模型（ML）；不做「個人化健康建議」（這超出本系列範圍）。

成功指標有兩個層次。系統層面：管線在 30 天內連續運行、沒有中斷；每日 SLI 報表都達到 Day 38 定義的 SLO（新鮮度 99%、完整性 99%、唯一性 100%）。使用者層面：Streamlit 儀表板每週至少有 5 個不重複瀏覽者（這個指標在 Day 44 部署後才有辦法量測）。

資料來源與合約

資料來源是行政院環境部「空氣品質監測網」，網址 https://airtw.moenv.gov.tw/。這個網站提供兩種取得方式：CSV 下載（https://data.gov.tw/dataset/40448 這個 dataset ID 在我們的參考實作為data.gov.tw 平台上的範例）與 JSON API。我們這次範例用 CSV 下載，因為它最容易重現（API 金鑰管理較複雜，留給讀者自行延伸）。授權為政府資料開放授權條款第 1 版。

原始資料的欄位大致如下：觀測時間（observed_at）、測站編碼（station_id）、測站名稱（station_name）、縣市（county）、AQI、PM2.5、PM10、O3、CO、SO2、NO2、溫度、濕度、風速。我們關心的核心欄位是 AQI 與 PM2.5，其他污染物與氣象欄位這次會落地但不進儀表板，留作 Day 45 之後的延伸應用。

把資料合約寫成 YAML，並沿用 Day 38 的 schema。這份合約會被 Day 43 的品質檢查、Day 44 的部署都用同一份讀取：

# de-journey/contracts/aqi_hourly.yml（與 Day 38 範例共用設定）
version: "1.0.0"
owner: data-platform@example.com
producer: 行政院環境部（airtw.moenv.gov.tw）
source_url: https://airtw.moenv.gov.tw/
license: 政府資料開放授權條款第 1 版
description: 全國測站每小時空氣品質量測，顆粒度為「一測站 × 一小時」。

schema:
  table: aqi.fct_aqi_hourly
  columns:
    - name: station_id
      type: string
      nullable: false
      unique_with: [observed_at]
      pattern: "^[A-Z0-9]{6,10}$"
    - name: observed_at
      type: timestamp
      nullable: false
    - name: aqi
      type: integer
      nullable: true
      range: { min: 0, max: 500 }
    - name: pm25
      type: double
      nullable: true
      range: { min: 0.0, max: 500.0 }
    - name: county
      type: string
      nullable: false
      values: ["臺北市","新北市","桃園市","臺中市","臺南市","高雄市",
               "基隆市","新竹市","新竹縣","苗栗縣","彰化縣","南投縣",
               "雲林縣","嘉義市","嘉義縣","屏東縣","宜蘭縣","花蓮縣",
               "臺東縣","澎湖縣","金門縣","連江縣"]
    - name: ingestion_at
      type: timestamp
      nullable: false

sli:
  freshness_hours: 26
  completeness_min: 0.99
  uniqueness_required: true

slo:
  freshness_pct: 0.99
  completeness_pct: 0.99
  uniqueness_pct: 1.00

這份合約是 Day 38 範例的擴充版，多了兩個關鍵欄位：第一，unique_with: [observed_at] 表示主鍵是複合（測站 + 觀測時間），這比單一欄位更貼近 OLAP 事實表的實務設計；第二，ingestion_at 是我們自己加的時間戳，記錄「這筆資料是何時被寫進倉儲的」，這對 freshness 計算很有用。除此之外，與 Day 38 相同的 SLO 門檻（freshness 99%、completeness 99%、uniqueness 100%）也沿用，這樣四天的監控指標有一致的基準。

維度模型：事實表與維度表

沿用 Day 23 與 Day 24 的星狀模型設計。我們的事實表是 fct_aqi_hourly，每一列代表「一個測站、一個小時、一組量測值」。顆粒度定義清楚後，下游所有彙總都會一致。圍繞事實表的有三張維度表：dim_stations（測站基本資料）、dim_counties（縣市維度）、dim_date（日期維度）。最後還有一張聚合寬表 mart_daily_summary，給月報與儀表板用。

維度模型是本專案的核心資產，所有 Day 41-45 的程式碼都引用同一組模型名稱：

模型	類型	顆粒度	更新頻率	用途
`stg_aqi_raw`	staging	測站 × 小時	每日	原始逐時量測，欄位型別尚未正規化
`stg_stations`	staging	測站	每週	測站基本資料（含縣市、地址）
`int_aqi_with_station`	intermediate	測站 × 小時	每日	事實表 join 維度表，補上縣市與測站名稱
`fct_aqi_hourly`	marts	測站 × 小時	每日	給儀表板與分析師的最終事實表
`dim_stations`	marts	測站	每週	測站維度（含縣市、地址、營運單位）
`dim_counties`	marts	縣市	不變	縣市維度（人口、面積、所在區域）
`mart_daily_summary`	marts	測站 × 日	每日	每日聚合（AQI 平均、最大、不良日）

這七個模型會在接下來四天陸續寫完。今天我們只寫兩個 staging 模型（stg_aqi_raw、stg_stations）與一個 intermediate 模型（int_aqi_with_station），其他四個會在 Day 42 與 Day 44 陸續補上。dbt 的好處是「模型相依由工具自動管理」，我們只負責寫 SELECT，剩下的交給 dbt。

為了讓「模型清單 → dbt 檔案 → 合約」三者永遠同步，我們寫一支「模型檔案清點」腳本，掃描 models/ 底下所有的 .sql 檔案，與 project_config.MODELS 對齊：

"""de-journey/scripts/check_models.py:確認 project_config.MODELS 與實際檔案一致。"""
from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "pipelines"))
from project_config import MODELS  # noqa: E402

MODELS_DIR = Path(__file__).resolve().parents[1] / "models" / "aqi_project" / "models"

declared = {m for layer in MODELS.values() for m in layer}
on_disk = {p.stem for p in MODELS_DIR.rglob("*.sql")}

missing = declared - on_disk
extra = on_disk - declared
if missing or extra:
    print("模型清單與檔案不同步：")
    for m in sorted(missing):
        print(f"  缺少檔案：{m}.sql")
    for m in sorted(extra):
        print(f"  多了檔案（請更新 project_config）：{m}.sql")
    sys.exit(1)
else:
    print(f"模型清單一致：{len(declared)} 個模型都在 {MODELS_DIR}")

輸出：

模型清單一致：3 個模型都在 models\aqi_project\models

這支腳本在 Day 42 之後特別有用：當我們補上 fct_aqi_hourly 等 marts 模型時，跑一次 check_models.py 就能立刻知道「我寫了新檔案但忘了更新 project_config.py」。CI 會把這支腳本放在「合約檢查」的工作裡，失敗就拒絕合併。

完整實作：建立 dbt 專案結構

專案結構沿用 Day 2 的工作目錄。我們在 de-journey/ 下新增 models/、pipelines/、dashboards/、tests/ 等目錄，並建立 dbt 專案骨架。整個過程大致 10 個步驟，可以用一支 bootstrap 腳本一次完成。

第一步：建立目錄結構與設定檔。

"""de-journey/scripts/init_project.py:Day 41 的專案初始化腳本。"""
from __future__ import annotations

from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]

# 1. 建立工作目錄結構
DIRS = [
    "data/raw", "data/processed",
    "warehouse",
    "models/aqi_project/models/staging",
    "models/aqi_project/models/intermediate",
    "models/aqi_project/models/marts",
    "models/aqi_project/tests",
    "models/aqi_project/macros",
    "pipelines",
    "dashboards",
    "logs",
    "contracts",
    "tests/e2e",
    ".github/workflows",
]
for d in DIRS:
    (ROOT / d).mkdir(parents=True, exist_ok=True)

# 2. 建立 dbt profiles 設定（指向 DuckDB）
profiles_path = Path.home() / ".dbt" / "profiles.yml"
profiles_path.parent.mkdir(parents=True, exist_ok=True)
profiles_path.write_text(
    "de_journey:\n"
    "  target: dev\n"
    "  outputs:\n"
    "    dev:\n"
    "      type: duckdb\n"
    "      path: ../warehouse/de-journey.duckdb\n"
    "      threads: 4\n",
    encoding="utf-8",
)
print("專案目錄與 dbt profiles 已建立")

這段把整個專案的目錄結構建立起來。目錄分得很細是有原因的：staging、intermediate、marts 三層模型分別放在不同目錄，是 dbt 官方推薦的標準結構（俗稱「dbt project structure」，參考 dbt Labs 的 how-we-structure-our-dbt-projects 文章）。pipelines/ 放 Python 腳本（ingest、transform、scheduler），dashboards/ 放 Streamlit，logs/ 放執行紀錄，contracts/ 放 Day 38 的合約檔。這套結構就是 Day 41-45 共用的工作目錄，後續三天不再變動。

第二步：建立 dbt 專案設定 dbt_project.yml。

# de-journey/models/aqi_project/dbt_project.yml
name: aqi_project
version: 1.0.0
profile: de_journey
model-paths: ["models"]
seed-paths: ["seeds"]
test-paths: ["tests"]
macro-paths: ["macros"]
target-path: "../../warehouse/dbt_target"
clean-targets:
  - "../../warehouse/dbt_target"
  - "dbt_packages"

models:
  aqi_project:
    staging:
      +materialized: view
      +schema: staging
    intermediate:
      +materialized: view
      +schema: intermediate
    marts:
      +materialized: table
      +schema: marts

這個設定把三層模型的 materialization 與 schema 分開。staging 與 intermediate 都是 view（不落地、不複製資料、隨時讀最新），marts 是 table（落地、查詢快、可以被外部 BI 工具直接讀）。target-path 寫相對路徑，這樣不同的開發者從各自的 clone 跑起來都會指向自己的 warehouse/。

第三步：定義 source。這是 dbt 對「原始資料」的描述，後續 staging 模型會從這裡讀：

# de-journey/models/aqi_project/models/staging/sources.yml
version: 2

sources:
  - name: aqi_raw
    description: 從環境部 CSV 落地進 DuckDB 的原始資料
    tables:
      - name: aqi_hourly
        description: 逐時空氣品質量測
        columns:
          - name: observed_at
            description: 觀測時間（台北時間）
          - name: station_id
            description: 測站編碼
          - name: aqi
            description: 空氣品質指標
          - name: pm25
            description: 細懸浮微粒濃度（μg/m³）
          - name: county
            description: 縣市名稱

sources.yml 是 dbt 的「資料來源契約」。下游 staging 模型用 {{ source('aqi_raw', 'aqi_hourly') }} 讀資料，這樣原始資料的欄位變動時，只要改 sources.yml 一個檔案就能反映到下游。

第四步：寫第一個 staging 模型 stg_aqi_raw。這個模型把原始欄位型別正規化，並過濾掉明顯錯誤的列：

-- de-journey/models/aqi_project/models/staging/stg_aqi_raw.sql
{{ config(materialized='view') }}

SELECT
    CAST(station_id AS VARCHAR)        AS station_id,
    CAST(observed_at AS TIMESTAMP)     AS observed_at,
    CAST(aqi AS INTEGER)               AS aqi,
    CAST(pm25 AS DOUBLE)               AS pm25,
    CAST(county AS VARCHAR)            AS county,
    CAST(ingestion_at AS TIMESTAMP)    AS ingestion_at
FROM {{ source('aqi_raw', 'aqi_hourly') }}
WHERE
    station_id IS NOT NULL
    AND observed_at IS NOT NULL
    AND aqi BETWEEN 0 AND 500
    AND (pm25 IS NULL OR pm25 BETWEEN 0 AND 500)

這是 dbt 模型的標準寫法。{{ config(...) }} 設定 materialization（覆寫專案預設）；{{ source(...) }} 讀上游 source；CAST 把字串轉成正確的型別；WHERE 子句先做基本的過濾。這個模型被 materialization 為 view（不複製資料），所以每次查詢都讀最新的原始表，Day 42 的 ingest 寫進新資料後，下游立刻看得到。

第五步：寫 stg_stations。這張維度表放測站基本資料。

-- de-journey/models/aqi_project/models/staging/stg_stations.sql
{{ config(materialized='view') }}

SELECT
    CAST(station_id AS VARCHAR)        AS station_id,
    CAST(station_name AS VARCHAR)      AS station_name,
    CAST(county AS VARCHAR)            AS county,
    CAST(address AS VARCHAR)           AS address,
    CAST(operator AS VARCHAR)          AS operator,
    CAST(latitude AS DOUBLE)           AS latitude,
    CAST(longitude AS DOUBLE)          AS longitude
FROM {{ source('aqi_raw', 'stations') }}
WHERE station_id IS NOT NULL

第六步：寫第一個 intermediate 模型 int_aqi_with_station。它把事實表與測站維度 join 起來，補上縣市與測站名稱：

-- de-journey/models/aqi_project/models/intermediate/int_aqi_with_station.sql
{{ config(materialized='view') }}

SELECT
    f.observed_at,
    f.station_id,
    s.station_name,
    f.county,
    f.aqi,
    f.pm25,
    f.ingestion_at
FROM {{ ref('stg_aqi_raw') }} f
    LEFT JOIN {{ ref('stg_stations') }} s
        ON f.station_id = s.station_id

{{ ref(...) }} 是 dbt 的「模型相依」寫法，比直接寫表名更穩：dbt 會自動建立 DAG，確保 stg_stations 在 int_aqi_with_station 之前執行；改模型名稱時只要改這裡，不用去找下游引用。LEFT JOIN 保證事實表的列數不會縮減（即使測站維度缺漏也保留事實）。

第七步：用 DuckDB 直接確認 dbt 跑得起來。

cd de-journey/models/aqi_project
uv run dbt deps        # 若有 packages.yml
uv run dbt run --select staging stg_aqi_raw stg_stations
uv run dbt run --select intermediate

輸出（簡化節錄）：

1 of 3 OK  created stg_aqi_raw  ............ [OK in 0.42s]
2 of 3 OK  created stg_stations  ........... [OK in 0.31s]
3 of 3 OK  created int_aqi_with_station  ... [OK in 0.18s]

這段是「Day 41 的可執行成果」：兩個 staging 模型、一個 intermediate 模型跑通。明天 Day 42 會加入 ingest 腳本（用真實或合成資料把這三張表填滿），後天 Day 43 會跑 dbt test 做品質檢查，大後天 Day 44 會部署到 GitHub Actions 與 Streamlit。

在收尾之前，我們寫一支「合約載入器」腳本，把 contracts/aqi_hourly.yml 載入並轉成 DuckDB 可讀的結構，Day 38 的合約檢查腳本會從這裡讀資料：

"""de-journey/scripts/load_contract.py:把 YAML 合約載入並轉成 Python 字典。"""
from __future__ import annotations

import sys
from pathlib import Path

try:
    import yaml
except ImportError:
    print("缺少 PyYAML，請跑：uv add pyyaml")
    sys.exit(1)

CONTRACT_PATH = Path(__file__).resolve().parents[1] / "contracts" / "aqi_hourly.yml"

contract = yaml.safe_load(CONTRACT_PATH.read_text(encoding="utf-8"))
print(f"已合約： {contract['schema']['table']}")
print(f"欄位數：{len(contract['schema']['columns'])}")
print(f"SLO：freshness={contract['slo']['freshness_pct']}, "
      f"completeness={contract['slo']['completeness_pct']}, "
      f"uniqueness={contract['slo']['uniqueness_pct']}")
# 輸出：
# 合約： aqi.fct_aqi_hourly
# 欄位數：6
# SLO：freshness=0.99, completeness=0.99, uniqueness=1.0

這支腳本做的事很單純：用 PyYAML 把 YAML 載入並印出關鍵欄位。實務上它會被 Day 38 的 check_contract.py 用 import 方式載入，而不是直接執行，這樣合約檔的所有欄位都能在 Python 程式碼裡直接用。請注意 try/except ImportError 是給讀者的友善提示：如果你忘了裝 PyYAML，腳本會告訴你怎麼裝，而不是丟一個看不懂的 stack。

共用設定檔：把「事實表與維度表清單」集中管理

為了確保 Day 41-45 的程式碼引用同一份模型清單，我們把模型清單寫成一個 Python 模組 pipelines/project_config.py。Day 42 的 ingest、Day 43 的測試、Day 44 的儀表板都從這裡 import 模型清單，不會出現「這個模型在 Day 41 叫 X、在 Day 44 叫 Y」這種前後不一致的狀況：

"""de-journey/pipelines/project_config.py:Day 41-45 共用的專案設定。"""
from __future__ import annotations

# 1. 資料來源（與 contracts/aqi_hourly.yml 對齊）
PROJECT = {
    "name": "aqi_pipeline",
    "owner": "data-platform@example.com",
    "license": "政府資料開放授權條款第 1 版",
    "source": {
        "name": "行政院環境部（airtw.moenv.gov.tw）",
        "url": "https://airtw.moenv.gov.tw/",
        "platform": "data.gov.tw",
    },
}

# 2. 模型清單（dbt models 與 DuckDB 視圖/表對應）
MODELS = {
    "staging": ["stg_aqi_raw", "stg_stations"],
    "intermediate": ["int_aqi_with_station"],
    "marts": ["fct_aqi_hourly", "dim_stations",
              "dim_counties", "mart_daily_summary"],
}

# 3. 指標定義（Day 43 評估、Day 44 儀表板都引用）
INDICATORS = {
    "national_daily_aqi_avg": "SELECT AVG(aqi) FROM fct_aqi_hourly WHERE DATE(observed_at) = ?",
    "bad_day_ratio": """
        SELECT CAST(SUM(CASE WHEN aqi > 100 THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*)
        FROM fct_aqi_hourly WHERE DATE(observed_at) = ?
    """,
    "monthly_pm25_avg": """
        SELECT AVG(pm25) FROM fct_aqi_hourly
        WHERE observed_at >= ? AND observed_at < ?
    """,
    "station_ranking": """
        SELECT station_id, AVG(aqi) AS avg_aqi
        FROM fct_aqi_hourly
        WHERE observed_at >= ?
        GROUP BY station_id ORDER BY avg_aqi DESC LIMIT 10
    """,
}

# 4. 排程設定（Day 42 APScheduler、Day 44 GitHub Actions 都用）
SCHEDULE = {
    "daily_run_cron": "0 3 * * *",          # 每日 03:00 (UTC+8) 跑 daily_run
    "station_refresh_cron": "0 4 * * 1",   # 每週一 04:00 跑測站基本資料
}

# 5. 告警（沿用 Day 38 的設計）
ALERT = {
    "channel": "logs",
    "breach_path": "logs/contract_breaches.jsonl",
}

print(f"專案：{PROJECT['name']}（{PROJECT['source']['name']}）")
print(f"模型：{len(MODELS['staging'])} staging + "
      f"{len(MODELS['intermediate'])} intermediate + "
      f"{len(MODELS['marts'])} marts")
# 輸出：
# 專案：aqi_pipeline（行政院環境部（airtw.moenv.gov.tw））
# 模型：2 staging + 1 intermediate + 4 marts

這個 project_config.py 是 Day 41-45 共用設定的單一來源（single source of truth）。後續四天的所有程式碼都 from project_config import MODELS, INDICATORS, SCHEDULE，這樣如果哪天要把模型改名、或調整指標定義，只要改這一個檔案。另一個好處是「程式碼意圖明確」：看到 INDICATORS["bad_day_ratio"] 就知道這是「不良日比例」指標，不需要再翻其他檔案。

最後寫一支「設定快照產生器」：當 Day 45 要做系列回顧時，這支腳本可以把當下的 project_config 印出來當作 commit message 的參考，證明「這四天我們的設定一直是這樣」。

"""de-journey/scripts/snapshot_config.py:把當下的共用設定印成文字檔。"""
from __future__ import annotations

import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "pipelines"))
from project_config import PROJECT, MODELS, INDICATORS, SCHEDULE, ALERT  # noqa: E402

ts = datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%d %H:%M:%S")
lines = [
    f"# project_config snapshot ({ts})",
    f"project: {PROJECT['name']}",
    f"source:  {PROJECT['source']['name']} ({PROJECT['source']['url']})",
    f"license: {PROJECT['license']}",
    "",
    "models:",
] + [f"  {layer}: {', '.join(names)}" for layer, names in MODELS.items()] + [
    "",
    "indicators:",
] + [f"  {name}: {sql.strip().splitlines()[0]}..." for name, sql in INDICATORS.items()] + [
    "",
    "schedule:",
] + [f"  {k}: {v}" for k, v in SCHEDULE.items()] + [
    "",
    "alert:",
] + [f"  {k}: {v}" for k, v in ALERT.items()]

snapshot = "\n".join(lines)
out_path = Path(__file__).resolve().parents[1] / "logs" / "project_config_snapshot.txt"
out_path.write_text(snapshot + "\n", encoding="utf-8")
print(f"已寫入 {out_path}")

這支腳本的價值在於「可審計」：每個階段的 commit 可以附上當下的設定快照，方便日後對照「這個指標是什麼時候改的、為什麼改」。實務上 Day 43 評估時也會用到：當指標異常時，把當下的快照與一個月前的快照比對，能立刻看出問題是「資料變了」還是「設定變了」。

為了一開始就能驗證工作目錄結構，我們再寫一支簡單的「目錄與檔案清點」腳本：列出所有 dbt 模型、scripts、pipelines 檔案，方便回頭檢查：

"""de-journey/scripts/list_assets.py:列出 Day 41 建立的資產。"""
from __future__ import annotations

from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]

for sub in ["models/aqi_project/models",
            "scripts", "pipelines", "contracts", "tests"]:
    files = sorted((ROOT / sub).rglob("*"))
    files = [f for f in files if f.is_file()]
    print(f"\n{sub}/：{len(files)} 個檔案")
    for f in files[:8]:
        print(f"  - {f.relative_to(ROOT)}")
    if len(files) > 8:
        print(f"  ... 其餘 {len(files) - 8} 個略")

輸出（節錄，實際檔案數會依後續 Day 42-44 補充而增加）：

models/aqi_project/models/：3 個檔案
  - stg_aqi_raw.sql
  - stg_stations.sql
  - int_aqi_with_station.sql
scripts/：4 個檔案
  - init_project.py
  - check_models.py
  - load_contract.py
  - snapshot_config.py
pipelines/：1 個檔案
  - project_config.py
contracts/：1 個檔案
  - aqi_hourly.yml

這支腳本純粹是「資產盤點」功能，沒有複雜邏輯，但當你接手別人的 repo 時，list_assets.py 是最快的「這個專案到底有什麼」入口。CI 也可以跑它，確保所有宣告的資產都實際在磁碟上。

常見錯誤與踩雷

第一個雷：先把整個管線架構想得太複雜。Day 41 的目的是「先把範圍與模型定下來」，不是把管線全部寫完。常見錯誤是當天就寫了 ingest、dbt、APScheduler、Streamlit、GitHub Actions 五個元件，結果後面三天沒東西可寫。建議 Day 41 只完成目錄結構、合約、staging 與 intermediate 三個模型。

第二個雷：合約欄位定義得太鬆（全部 nullable、沒有值域）。Day 38 教過的「值域」「列舉」這兩個約束，是抓「數字看起來對、其實錯了」這類錯誤的利器。Day 41 寫合約時就應該把約束填好，Day 43 的測試才有東西可以驗。

第三個雷：模型命名不一致。Day 41 寫 stg_aqi_raw、Day 44 寫 fct_aqi_raw，前後讀者會搞不清楚哪個是事實表、哪個是 staging。請把模型清單寫在 project_config.py 一個地方，所有程式碼從那裡 import。

第四個雷：忘了設定 dbt 的 profile 路徑。Day 2 已經示範過 ~/.dbt/profiles.yml，但 Day 41 的同事如果換了一台機器，忘記建這個檔，dbt run 就會噴 Connection test failed。請在 scripts/init_project.py 裡把 profiles.yml 自動寫進去（像我們今天示範的）。

第五個雷：把 source 與 staging 混在一起。source 描述「原始資料的長相」（CSV 落地後長什麼樣），staging 描述「正規化後的長相」（型別正確、欄位命名統一）。兩者分開才能在原始資料變動時，只改 source 不改 staging。

效能與實務提醒

dbt 的 staging 用 view、intermediate 用 view、marts 用 table，是 2025 年 dbt 社群的主流配置。view 不複製資料、隨時讀最新、缺點是每次查詢都要重新計算；table 落地、查詢快、缺點是資料會延遲（要等 dbt run 跑完才會更新）。我們這個專案的 staging 與 intermediate 都是 view，因為它們只是「欄位型別轉換」與「join」，沒什麼運算成本；marts 用 table，是因為儀表板要讀且要快。

實務上有三個取捨值得記得。第一，model 名稱長度：dbt 模型名稱越短越好，但要能表達意圖；fct_aqi_hourly 是好名稱，f_a_h 雖然短但讀者要花時間猜。第二，schema 命名：用 schema 把 staging/intermediate/marts 三層分開（staging.stg_aqi_raw、marts.fct_aqi_hourly），BI 工具就能用 schema 過濾、只讀 marts。第三，ref 的顆粒度：每個模型只依賴直接上游，不要跨層 ref；marts.fct_aqi_hourly 應該 ref intermediate.int_aqi_with_station，而不是直接 ref stg.stg_aqi_raw。

另一個工程上的提醒：DuckDB 在 dbt run 時，會把每個模型編譯成原生程式碼執行，速度比多數 OLAP 引擎快。當 staging 模型很多（10 個以上）時，dbt run 可能要跑幾分鐘；可以用 --select +fct_aqi_hourly 只跑上游必要模型，縮短開發迭代時間。Day 42 會把這個技巧接進排程器，只跑必要的模型。

小結

今天把專案篇的「地基」蓋起來。我們定義了專案範圍（每天自動抓環境部空氣品質資料、產出維度模型、用 Streamlit 儀表板呈現）、共用合約 aqi_hourly.yml（沿用 Day 38 結構）、七個 dbt 模型（含 staging、intermediate、marts 三層），以及 project_config.py 這個「Day 41-45 共用設定」的單一入口。今天完成的程式：3 個 dbt 模型（stg_aqi_raw、stg_stations、int_aqi_with_station）跑通，加上完整的工作目錄結構與合約檔。重點觀念有三：第一，先定義範圍再寫程式；第二，把合約與模型清單集中在一個地方管理；第三，staging 用 view、marts 用 table 的 materialization 策略是 2025 年的主流配置。

明天，我們會進入專案篇的第二天「管線實作與排程」。我們會把 ingest 腳本（從環境部抓資料）寫完、用 dbt run 把 marts 補上、用 APScheduler 或 GitHub Actions 設定每日 03:00 的自動執行，並把 Day 38 的合約檢查接進排程。今天的 project_config.py 會被大量引用：明天所有的程式碼會 from project_config import MODELS, INDICATORS, SCHEDULE，這就是 Day 41-45 共用設定的具體表現。

結語

今天的重點是「先把專案想清楚再動手」。我們把過去 40 天的工具鏈（uv、DuckDB、Polars、dbt、APScheduler、Streamlit）組合成一個可運行的專案骨架，並定義好合約、模型與共用設定。讀完這篇你應該能回答：為什麼 Day 41 還沒開始寫 ingest 就要先定合約？七個模型的 staging/intermediate/marts 三層怎麼分工？project_config.py 為什麼是 Day 41-45 共用設定的單一來源？明天，我們會把這套骨架補上「採集與排程」，讓管線真的每天自動跑起來。

延伸資源

行政院環境部空氣品質監測網：https://airtw.moenv.gov.tw/。本專案的唯一原始資料來源，採政府資料開放授權條款第 1 版。
政府資料開放平臺：https://data.gov.tw/。台灣政府開放資料的統一入口。
dbt Labs「How we structure our dbt projects」：https://docs.getdbt.com/best-practices/how-we-structure/。本系列採用的 staging/intermediate/marts 三層結構以此為基礎。
Ralph Kimball《The Data Warehouse Toolkit》第三版（2013）：星狀模型、維度表、SCD 的經典教科書，Day 23 與 Day 24 也引用過。
Day 38 章節（監控與資料合約）：今天這份合約是 Day 38 的延伸，把 Day 38 的 SLI／SLO 設計落到具體欄位。

Hao程式筆記

搜尋此網誌