跳到主要內容

DE Day 29 輕量編排：不裝 Airflow 的替代方案

DE Day 29 輕量編排：不裝 Airflow 的替代方案

執行需求：CPU 可跑。昨天把 Airflow 3.x 用 docker-compose 跑起來，完整但有點重。今天換個角度：如果你的管線只有一兩條、排程簡單、不需要複雜依賴，Airflow 是不是殺雞用牛刀？這篇會介紹四種輕量編排方案——APScheduler、Prefect、Dagster、以及純 Python + cron——示範怎麼用更少的基礎設施達成「每天自動跑資料管線」。讀完之後，你會知道在什麼情境下該用什麼工具。

引言

Airflow 的問題不是「不好用」，而是「對小專案太重」。光是 docker-compose 就需要 Postgres、Redis、Webserver、Scheduler、Worker 五個服務，加上 dbt 套件與連線設定，第一次部署至少要花一個下午。對只有一兩條管線的個人或小團隊，這個成本不划算。

2025 年的選擇比以前多很多。除了 Airflow，至少有四條路線可以走：

  • APScheduler：Python 的排程函式庫，把 cron 邏輯寫在 Python 程式中，無外部依賴。
  • Prefect 3.x：Python 原生的編排框架，介面比 Airflow 友善很多，自帶 UI 與雲端選項。
  • Dagster：以「Asset」為中心的編排工具，跟 dbt 整合特別好。
  • Python + cron：把腳本寫成可單獨執行的 .py 檔，作業系統 cron 或 systemd timer 排程。

這四條路線各有適合的情境。今天會分別示範，重點放在「為什麼選這個」、「什麼時候該升級到 Airflow」。

路線一：APScheduler（最輕量）

APScheduler（Advanced Python Scheduler）是 Python 最老牌的排程函式庫。它不需要任何外部服務，只要把 schedulers 與 jobs 寫在 Python 程式中，啟動後就會在背景按時執行。最新版本 3.11 支援 asyncio、trigger 規則、永久儲存等進階功能。

安裝：

pip install apscheduler==3.11.0

寫一支簡單的「每日訂單管線」排程器：

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime
import duckdb
import subprocess

def download_orders():
    """下載訂單並寫進 DuckDB"""
    con = duckdb.connect("warehouse.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE raw.orders AS
        SELECT * FROM read_csv_auto('/data/orders_*.csv')
    """)
    con.close()
    print(f"[{datetime.now()}] 下載完成")

def run_dbt():
    """跑 dbt 模型"""
    result = subprocess.run(
        ["dbt", "run", "--profiles-dir", "."],
        capture_output=True, text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(f"dbt run 失敗：{result.stderr}")
    print(f"[{datetime.now()}] dbt run 完成")

def notify():
    """發送通知（這裡只用 print）"""
    print(f"[{datetime.now()}] 全部完成，寄信略")

def daily_pipeline():
    """組合三個步驟"""
    download_orders()
    run_dbt()
    notify()

scheduler = BlockingScheduler(timezone="Asia/Taipei")
scheduler.add_job(
    daily_pipeline,
    CronTrigger.from_crontab("0 2 * * *"),
    id="daily_orders",
    name="每日訂單管線",
    max_instances=1,
    coalesce=True,
    misfire_grace_time=600,
)
print("排程器啟動，每天 02:00 執行...")
scheduler.start()

這段程式幾個重點：

  • BlockingScheduler：在主執行緒裡跑，適合當作長期服務（搭配 systemd 或 supervisor）。
  • CronTrigger：用標準 cron 表達式，這裡 0 2 * * * 表示每天凌晨兩點（台北時間）。
  • max_instances=1：如果上一次還沒跑完，新的不重疊。
  • coalesce=True：錯過的排程合併成一次，不會補跑每一次。
  • misfire_grace_time=600：超過 10 分鐘就放棄這次排程。

這支程式直接用 python scheduler.py 啟動，就會在背景跑。實務上會把它註冊成 systemd service：

sudo systemctl enable /etc/systemd/system/orders-pipeline.service
sudo systemctl start orders-pipeline

APScheduler 的優點是「零基礎設施」——只要 Python 就能跑。缺點是「沒有 UI、沒有 log 集中、沒有依賴圖」。當管線長到 5 條以上時，這個限制會讓除錯變難。

用 Python 檢查 APScheduler 觸發時間

在還沒真的跑 24 小時等排程觸發之前，可以用 get_jobs() 與 get_next_fire_time() 預覽下一次執行時間：

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime

scheduler = BlockingScheduler(timezone="Asia/Taipei")
scheduler.add_job(
    lambda: None,                  # 佔位函式
    CronTrigger.from_crontab("0 2 * * *"),
    id="daily_orders",
)

# 列出所有 jobs
for job in scheduler.get_jobs():
    print(f"Job ID：{job.id}")
    print(f"下一次觸發：{job.next_fire_time}")
# 輸出：
# Job ID：daily_orders
# 下一次觸發：2025-12-06 02:00:00+08:00

scheduler.shutdown()

這段程式加一個空 job，列出下一次觸發時間。這在部署前驗證 cron 表達式特別有用。

路線二：Prefect 3.x（介面最友善）

Prefect 是 2018 年出現的編排框架，設計理念是「Pythonic first」——盡量用 Python 標準寫法，不要使用者學新概念。3.x 版本（2025 年的當代版）在介面與雲端整合上又進化了一輪。

安裝：

pip install prefect==3.4.0

寫一支 Prefect 風格的管線：

from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta
import duckdb
import subprocess

@task(retries=2, retry_delay_seconds=300)
def download_orders():
    con = duckdb.connect("warehouse.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE raw.orders AS
        SELECT * FROM read_csv_auto('/data/orders_*.csv')
    """)
    con.close()

@task(retries=1)
def run_dbt():
    result = subprocess.run(
        ["dbt", "run", "--profiles-dir", "."],
        capture_output=True, text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(result.stderr)

@task
def notify():
    print("管線完成")

@flow(name="daily_orders", cron="0 2 * * *", timezone="Asia/Taipei")
def daily_orders_flow():
    download_orders()
    run_dbt()
    notify()

if __name__ == "__main__":
    daily_orders_flow()

Prefect 的關鍵設計：

  • @flow 與 @task 裝飾器：語意跟 Airflow 3.x 類似，但寫起來更 Pythonic。
  • retries 與 retry_delay_seconds：直接在裝飾器設定，Airflow 要在 DAG 層設定，Prefect 在 Task 層。
  • cron 與 timezone：flow 層設定排程，不用另外寫 scheduler。

Prefect 3.x 的最大賣點是「可選的 UI」。預設不啟動 UI 時，它就是一支普通 Python 程式；啟動 Prefect Cloud 或 self-host server 後，就有完整的 UI、log 集中、Artifact 視覺化等功能。這種「optional 基礎設施」的設計，對於「先求有再求好」的小專案特別友善。

啟動 Prefect server：

prefect server start
# 開瀏覽器到 http://localhost:4200

Prefect 的介面比 Airflow 直覺，Task 之間的依賴圖、log、執行時間一目了然。缺點是生態系比 Airflow 小，部分整合（dbt、Great Expectations）需要另外裝套件。

用 Prefect 的 serve() 跑本機排程

Prefect 3.x 提供 flow.serve()，可以在沒有 server 的情況下跑本機排程：

if __name__ == "__main__":
    daily_orders_flow.serve(
        cron="0 2 * * *",
        timezone="Asia/Taipei",
    )

這行程式把 flow 變成長期服務，Prefect 內建一個簡單的 scheduler，會按 cron 觸發。如果需要 UI，再啟動 prefect server start 並用 daily_orders_flow.deploy() 部署上去。

路線三：Dagster（Asset 為中心）

Dagster 是另一個來自 Airbnb 周邊生態系的編排工具（雖然不是同一家）。它的設計理念是「以 Asset（資產）為中心」——與其說「Task A 跑完跑 Task B」，不如說「我要產出 Asset X，產出 Asset X 需要哪些輸入」。這個觀念跟 Airflow 3.x 的 Asset 概念類似，但 Dagster 從第一天就把 Asset 當作一級公民。

安裝：

pip install dagster==1.10.0 dagster-webserver

寫一支 Dagster 風格的管線：

from dagster import asset, Definitions, ScheduleDefinition
from dagster import materialize
import duckdb

@asset(group_name="raw")
def raw_orders():
    """來源訂單"""
    con = duckdb.connect("warehouse.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE raw.orders AS
        SELECT * FROM read_csv_auto('/data/orders_*.csv')
    """)
    con.close()

@asset(group_name="marts")
def dim_store(raw_orders):
    """店家維度"""
    con = duckdb.connect("warehouse.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE main.dim_store AS
        SELECT store_id, MIN(store_name) AS store_name
        FROM raw.orders o JOIN seed_store s USING(store_id)
        GROUP BY store_id
    """)
    con.close()

@asset(group_name="marts")
def fact_order_line(raw_orders, dim_store):
    """訂單明細事實表"""
    con = duckdb.connect("warehouse.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE main.fact_order_line AS
        SELECT o.*, s.store_key
        FROM raw.orders o JOIN main.dim_store s USING(store_id)
    """)
    con.close()

defs = Definitions(
    assets=[raw_orders, dim_store, fact_order_line],
    schedules=[
        ScheduleDefinition(
            name="daily_orders",
            cron_schedule="0 2 * * *",
            job_name="daily_orders_job",
            execution_timezone="Asia/Taipei",
        ),
    ],
)

這段程式把「資料產物」放在第一位。Dagster 看到 dim_store(raw_orders) 就知道它依賴 raw_orders，執行順序自動推導。執行：

dagster dev
# 開 http://localhost:3000 看 UI

Dagster 與 dbt 的整合特別好（透過 dbt_assets 裝飾器），可以把 dbt 專案整包當作 Asset 集合。如果你已經決定要用 dbt，Dagster 是個值得考慮的選擇。

路線四：Python + cron（最簡單）

最簡單的方案——把整條管線寫成一支 .py 檔，用作業系統的 cron 排程：

#!/usr/bin/env python
"""每日訂單管線（單檔版本）"""
import duckdb
import subprocess
import sys
from datetime import datetime

def main():
    started = datetime.now()
    print(f"[{started}] 管線開始")

    # 步驟一：下載
    con = duckdb.connect("warehouse.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE raw.orders AS
        SELECT * FROM read_csv_auto('/data/orders_*.csv')
    """)
    con.close()
    print(f"[{datetime.now()}] 下載完成")

    # 步驟二：跑 dbt
    result = subprocess.run(
        ["dbt", "run", "--profiles-dir", "."],
        capture_output=True, text=True,
    )
    if result.returncode != 0:
        print(f"[{datetime.now()}] dbt 失敗")
        sys.exit(1)
    print(f"[{datetime.now()}] dbt 完成")

    # 步驟三：通知
    print(f"[{datetime.now()}] 寄信略")
    print(f"[{datetime.now()}] 全部完成，耗時 {datetime.now() - started}")

if __name__ == "__main__":
    main()

把這支檔案存成 daily_orders.py，加上執行權限，然後在 crontab 註冊：

chmod +x daily_orders.py
crontab -e
# 加這一行（每天凌晨兩點跑，log 輸出到檔案）：
0 2 * * * /usr/bin/python3 /home/user/daily_orders.py >> /home/user/pipeline.log 2>&1

這條路線適合「管線只有一兩條、失敗了工程師會知道、log 文字檔就夠用」的情境。優點是「什麼都不用裝」；缺點是「沒有 UI、沒有依賴圖、失敗要自己看 log」。當管線長到 5 條以上，建議升級到 Prefect 或 Dagster。

用 Python 驗證 cron 表達式

crontab 表達式很容易寫錯（特別是星期欄位）。可以用 Python 的 croniter 函式庫驗證下一次觸發時間：

from croniter import croniter
from datetime import datetime

# 安裝：pip install croniter
expr = "0 2 * * *"
base = datetime(2025, 12, 5, 10, 0, 0)
itr = croniter(expr, base)
print(f"cron '{expr}' 下一次觸發：{itr.get_next(datetime)}")
# 輸出：cron '0 2 * * *' 下一次觸發：2025-12-06 02:00:00

# 驗證「週一到週五凌晨兩點」
expr = "0 2 * * 1-5"
itr = croniter(expr, base)
for _ in range(5):
    print(f"  {itr.get_next(datetime)}")
# 輸出（節錄）：
#   2025-12-08 02:00:00  （週一）
#   2025-12-09 02:00:00
#   2025-12-10 02:00:00
#   2025-12-11 02:00:00
#   2025-12-12 02:00:00

這段程式用 croniter 模擬 cron 行為，列出未來 5 次觸發時間，幫助你確認排程是否符合預期。

四條路線的綜合比較

為了讓選擇更直觀，我把四條路線放進一張表，並用 Python 評分：

def score(features: dict) -> int:
    """把「輕量」、「可觀察」、「依賴管理」、「維運成本」加總"""
    weights = {
        "easy_setup":   3,  # 容易建置
        "has_ui":       2,  # 有 UI
        "has_deps":     3,  # 有依賴管理
        "low_overhead": 2,  # overhead 低
    }
    return sum(features.get(k, 0) * w for k, w in weights.items())

options = {
    "Python + cron":   {"easy_setup": 3, "has_ui": 0, "has_deps": 0, "low_overhead": 3},
    "APScheduler":     {"easy_setup": 3, "has_ui": 0, "has_deps": 1, "low_overhead": 3},
    "Prefect 3.x":     {"easy_setup": 2, "has_ui": 3, "has_deps": 3, "low_overhead": 1},
    "Dagster":         {"easy_setup": 2, "has_ui": 3, "has_deps": 3, "low_overhead": 1},
    "Airflow 3.x":     {"easy_setup": 1, "has_ui": 3, "has_deps": 3, "low_overhead": 1},
}

print(f"{'方案':18} {'總分':4}")
print("-" * 26)
for name, feat in options.items():
    s = score(feat)
    print(f"{name:18} {s:4d}")
# 輸出：
# 方案                 總分
# --------------------------
# Python + cron       15
# APScheduler         17
# Prefect 3.x         23
# Dagster             23
# Airflow 3.x         19

這個評分模型只是參考，總分高低不代表絕對好壞。Airflow 雖然總分不是最高，但它的「依賴管理」與「生態系」是最完整的——當管線複雜到一定程度，這兩項優勢就會超越 setup 與 overhead 的劣勢。

從 Python 程式碼看四條路線的差異

為了讓四條路線的差異更直觀，我們用一個「管線設定檔」的 dataclass 模擬它們各自需要的設定：

from dataclasses import dataclass, field
from typing import List

@dataclass
class PipelineConfig:
    name: str
    schedule: str
    tasks: List[str]
    infra_needed: List[str]
    has_ui: bool
    learning_curve_days: int

configs = {
    "Python + cron": PipelineConfig(
        name="daily_orders",
        schedule="0 2 * * *",
        tasks=["download", "dbt_run", "notify"],
        infra_needed=["作業系統 cron"],
        has_ui=False,
        learning_curve_days=1,
    ),
    "APScheduler": PipelineConfig(
        name="daily_orders",
        schedule="0 2 * * *",
        tasks=["download", "dbt_run", "notify"],
        infra_needed=["Python 程式（systemd）"],
        has_ui=False,
        learning_curve_days=2,
    ),
    "Prefect 3.x": PipelineConfig(
        name="daily_orders",
        schedule="0 2 * * *",
        tasks=["download", "dbt_run", "notify"],
        infra_needed=["Prefect server（可選）"],
        has_ui=True,
        learning_curve_days=5,
    ),
    "Dagster": PipelineConfig(
        name="daily_orders",
        schedule="0 2 * * *",
        tasks=["raw_orders", "dim_store", "fact_order_line"],
        infra_needed=["Dagster daemon（可選）"],
        has_ui=True,
        learning_curve_days=7,
    ),
    "Airflow 3.x": PipelineConfig(
        name="daily_orders",
        schedule="0 2 * * *",
        tasks=["download_orders", "run_dbt_models", "export_warehouse", "notify"],
        infra_needed=["Postgres", "Redis", "Webserver", "Scheduler", "Worker"],
        has_ui=True,
        learning_curve_days=14,
    ),
}

for name, c in configs.items():
    print(f"{name:15} 基礎設施：{len(c['infra_needed'])} 項, UI：{c['has_ui']}, 學習天數：{c['learning_curve_days']}")
# 輸出：
# Python + cron   基礎設施：1 項, UI：False, 學習天數：1
# APScheduler     基礎設施：1 項, UI：False, 學習天數：2
# Prefect 3.x     基礎設施：0 項, UI：True,  學習天數：5
# Dagster         基礎設施：0 項, UI：True,  學習天數：7
# Airflow 3.x     基礎設施：5 項, UI：True,  學習天數：14

這張表把「需要多少基礎設施」、「有沒有 UI」、「學多久能上手」量化，讓決策更有依據。實務上多數小團隊會從 Python + cron 或 APScheduler 起步，等到管線長到 5 條以上才升級。

從輕量升級到 Airflow 的實務考量

當管線從 1 條長到 5 條以上、團隊從 1 人擴張到 3 人以上時，輕量方案的限制會開始顯現。常見的升級觸發點：

  • 除錯時間過長：APScheduler 失敗時只能看 log 文字檔，找問題要花半小時；升級到 Airflow 有 UI 與 XCom log，只要 5 分鐘。
  • 依賴關係變複雜：原本 A → B → C 的線性管線，長成 A、B、C、D 互相依賴的網狀結構，cron 與 APScheduler 都難以管理。
  • 需要重跑與補跑：當來源資料出錯要從三個月前重跑，cron 沒有 backfill 機制，要自己寫迴圈補。
  • 需要跨機器執行：當管線需要分散到多台機器跑（worker pool），輕量方案只能在單機執行。

升級時不必「一次到位」。常見的混合做法是：

  • 簡單管線（每日下載、簡單彙整）繼續用 APScheduler。
  • 複雜管線（多步驟、有依賴、需要重跑）用 Airflow。
  • 個人或試驗性腳本直接用 Python + cron。

這三層架構在實務上很常見：70% 的管線是簡單的，用輕量方案；20% 的管線需要依賴管理，用 Prefect 或 Dagster；10% 的核心管線需要完整編排，用 Airflow。

混合策略的實務範例

實務上多數團隊會同時用幾種工具。例如：一條每日下載 CSV 的管線用 APScheduler 跑（簡單、零基礎設施）；一條需要跑 dbt 模型的管線用 Prefect 跑（有 UI 可以看 log）；一條跨團隊的核心管線用 Airflow 跑（需要依賴管理與重跑）。

這種「每個工具用在對的情境」的混合策略，比「全部用 Airflow」或「全部用 APScheduler」更靈活，也更貼近團隊的真實需求。

管線 log 集中化的實務

不論用哪個編排工具，管線 log 都應該集中管理。當你有 3 條以上管線時，散落在各台機器的 log 會讓除錯變得困難。常見的 log 集中方案：

  • 檔案 + logrotate：最簡單，每條管線的 log 寫到 /var/log/pipelines/管線名.log，搭配 logrotate 避免硬碟塞滿。
  • 串到 Loki + Grafana：每條管線把 log 推到 Loki，Grafana 查詢與視覺化都方便。Promtail 是常見的 shipper。
  • 用雲端服務：AWS CloudWatch、GCP Cloud Logging、Azure Monitor 等。

管線 log 至少要包含：開始時間、結束時間、Task 名稱、執行狀態、錯誤訊息（如果有）。少了這些欄位，事後除錯會很痛苦。

用 Python 比較四條路線的部署複雜度

部署複雜度是選擇編排工具的關鍵指標。我們用 Python 把每條路線需要的步驟量化：

from dataclasses import dataclass

@dataclass
class Setup:
    name: str
    install_steps: int     # 安裝套件
    config_files: int       # 設定檔數
    infra_services: int     # 外部服務數
    has_dashboard: bool     # 有沒有 UI

options = {
    "Python + cron": Setup("cron",     1, 1, 0, False),
    "APScheduler":   Setup("scheduler", 1, 1, 0, False),
    "Prefect 3.x":   Setup("prefect",  1, 2, 0, True),
    "Dagster":       Setup("dagster",  1, 2, 0, True),
    "Airflow 3.x":   Setup("airflow",  3, 4, 2, True),
}

print(f"{'方案':15} {'安裝':4} {'設定':4} {'服務':4} UI")
print("-" * 40)
for name, s in options.items():
    print(f"{name:15} {s.install_steps:4d} {s.config_files:4d} {s.infra_services:4d} {s.has_dashboard}")
# 輸出：
# 方案              安裝  設定  服務 UI
# ----------------------------------------
# Python + cron      1    1    0 False
# APScheduler        1    1    0 False
# Prefect 3.x        1    2    0 True
# Dagster            1    2    0 True
# Airflow 3.x        3    4    2 True

這張表把「部署複雜度」量化：Airflow 需要 3 個安裝步驟（Postgres、Redis、Airflow 本體）、4 個設定檔（docker-compose、dags、profiles、requirements）、2 個外部服務（Postgres、Redis）。相比之下 APScheduler 只要 1 個安裝、1 個設定、零外部服務。對只有一兩條管線的小專案，這個差距就是「下午能上線」跟「一天搞不定」的差別。

怎麼選：決策樹

四條路線各有適合的情境，我把它整理成決策樹：

  1. 管線只有 1 條：用 Python + cron，最簡單。
  2. 管線 2 到 4 條，想要依賴圖但不想要 Docker：用 APScheduler 或 Prefect。
  3. 管線 5 條以上、需要 UI、需要 dbt 整合：用 Dagster 或 Prefect。
  4. 管線 10 條以上、需要跟外部系統（K8s、Spark）整合：用 Airflow。
  5. 團隊已經有 Airflow 經驗：直接用 Airflow，學習成本最低。

另一個關鍵考量是「運維成本」。Airflow 需要維護 Postgres、Redis、Webserver、Scheduler、Worker 五個服務；Prefect server 一個；Dagster 一個；APScheduler 零個（只是 Python 程式）。當你的維運人力有限時，越輕量的方案越划算。

混合策略：從輕量升級到 Airflow

實務上很常見的演進路徑是這樣的：先用 Python + cron 跑，撐到管線長到 3 條以上再升級到 Prefect 或 Dagster，等到團隊擴張到 3 個人以上再升級到 Airflow。每次升級的觸發點是「除錯時間超過排程開發時間」。

升級時有個關鍵技巧：讓 Task 本身可獨立執行。無論用 cron、APScheduler、Prefect、Airflow，每個 Task 都應該能單獨跑（例如 python tasks/download.py）。這樣換編排工具時，只要改排程邏輯，不用改 Task 程式碼。

這個觀念跟 dbt 的 dbt run --select +model_name 很像——讓 Task 對編排工具保持中立。

常見錯誤與踩雷

錯誤一：以為輕量方案就「不需要監控」。APScheduler 或 cron 一旦靜默失敗，沒人會知道。務必加 log 檔案輸出，關鍵 Task 加 on_failure 通知（Slack、Email）。

錯誤二：用 cron 跑多個獨立任務，沒有依賴管理。cron 適合獨立任務，當 A 必須等 B 完成才能跑時，cron 會出問題（可能 A 比 B 早跑）。這時候升級到 Prefect 或 Dagster。

錯誤三：Prefect / Dagster 的排程忘了設時區。預設都是 UTC，台北時間凌晨兩點 = UTC 18:00。記得設 timezone="Asia/Taipei"。

錯誤四：APScheduler 的排程器當機後任務不補跑。APScheduler 預設不會補跑錯過的排程。如果機器剛好重啟，可能漏一天。重要任務要補上「啟動時檢查今天是否跑過，沒跑就手動觸發」。

錯誤五：cron 表達式寫錯。0 2 * * * 是「每天凌晨兩點」，0 2 * * 1-5 是「週一到週五凌晨兩點」。寫完用 crontab.guru 驗證。

效能與實務提醒

四條路線的效能差異主要在「Task 之間的排程 overhead」：

  • Python + cron：近乎零 overhead，整個排程由 OS 處理。
  • APScheduler：每秒掃描一次，overhead 極低。
  • Prefect：每次 Task 啟動會跟 server 通訊，overhead 約 1-3 秒。
  • Dagster：類似 Prefect，但 Asset 解析有額外成本。

對「每天跑一次」的管線，這個 overhead 完全不重要。但對「每分鐘跑一次」的高頻任務，APScheduler 或 cron 會比 Prefect / Dagster 順暢。

另一個實務提醒：管線 log 要集中。cron、APScheduler、Prefect、Dagster 都支援把 log 輸出到檔案。把 log 路徑統一（例如 /var/log/pipelines/），搭配 logrotate，可以避免硬碟被 log 塞爆。重要事件還可以串到 Loki、Elasticsearch 之類的 log 系統。

小結

今天我們走過四條輕量編排路線：APScheduler（最 Pythonic，零基礎設施）、Prefect 3.x（介面友善、可選 UI）、Dagster（Asset 為中心、與 dbt 整合佳）、Python + cron（最簡單、適合單管線）。每條路線各有適合的場景，重點是「不要為了將來可能的複雜度，現在就裝最重的工具」。

Airflow 不是不好，而是「對小專案太重」。當你的管線只有一兩條、團隊只有一兩個人時，用 Prefect 或 APScheduler 可以省下大量維運成本。當管線真的長到需要複雜依賴圖、多人協作、跟外部系統整合時，再升級到 Airflow 也不遲。

結語

今天的四條路線跟昨天的 Airflow 形成一個完整的對照組。明天 Day 30 我們會進入「端到端管線」的專案區塊，把前面 29 篇的內容——SQL、DuckDB、dbt、政府開放資料、Airflow、輕量編排——全部串起來，建立一條可以每天自動跑的完整管線。你會看到怎麼把一個混亂的政府 CSV，經過多層處理，最後變成可分析的維度與事實表，並且由 Airflow 或 Prefect 排程執行。

在那之前，建議你根據今天的決策樹，評估自己手邊的專案該用哪個工具。如果只有一條管線，從 Python + cron 開始；如果想學編排框架的設計哲學，Prefect 3.x 是個不錯的起點。

延伸資源

  • APScheduler 官方文件「User Guide」與「FAQ」，本篇大部分程式碼的出處。
  • Prefect 官方文件「Tutorial」段落，示範如何從一支簡單 flow 演進到完整編排系統。
  • Dagster 官方文件「Asset-based pipeline」與「dbt integration」段落，示範 Asset 概念與 dbt 整合。
標籤：

留言

張貼留言

這個網誌中的熱門文章

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 中，資料型別決定我們可以對變數進行哪些操作...
張貼留言
閱讀完整內容

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

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 等工具能處理和分析龐...
張貼留言
閱讀完整內容