k8s-airflow/01-airflow功能介紹.md

Apache Airflow 介紹

---

1. 什麼是 Apache Airflow？

• Apache Airflow 是「用程式碼管理所有資料流程(Configuration as Code)的平台」

• 適用於 ETL、資料整合、報表產生、機器學習前後處理等流程。
• 使用者透過 Python 程式碼 來 編寫、排程 和監控 數據工作流程。

核心組成

Airflow 的運作邏輯是由以下三個核心概念組成的：

1. 流程怎麼走 : DAG (有向無環圖)：
   • 這是任務流程的「藍圖」。所有的任務順序、依賴關係（先做 A，再做 B）都在這裡定義。
2. 工作要怎麼做 : Operator (操作員)：
   • 定義任務「實際做什麼」的模板。
   • Providers：Airflow 內建 AWS, GCP, Snowflake 等數千種現成的連接器，這是傳統工具較難追上的生態優勢。
3. 實際跑出來的結果 : Task (任務)：
   • 當 DAG 被執行時，Operator 就會被實例化成為一個 Task，並擁有自己的狀態（排程中、執行中、成功、失敗）。

from airflow import DAG
from airflow.decorators import task
from datetime import datetime
import json
from pathlib import Path

with DAG(
    dag_id="taskflow_realistic_etl_path_xcom",
    start_date=datetime(2024, 1, 1),
    # schedule="0 2 * * *",
    catchup=False,
    tags=["example", "taskflow", "xcom"],
) as dag:

    @task
    def extract() -> str:
        # 實務上：抓 API/DB 後落地成檔案（回傳「路徑」最推薦）
        out = Path("/tmp/airflow_demo_extract.json")
        out.write_text(json.dumps({"rows": 3, "data": [1, 2, 3]}))
        return str(out)  # 這個字串會自動進 XCom

    @task
    def transform(path: str) -> str:
        p = Path(path)
        payload = json.loads(p.read_text())

        # 假裝做轉換：把每個值 * 10
        payload["data"] = [x * 10 for x in payload["data"]]

        out = Path("/tmp/airflow_demo_transform.json")
        out.write_text(json.dumps(payload))
        return str(out)

    @task
    def load(path: str) -> None:
        payload = json.loads(Path(path).read_text())
        # 實務上：寫入 Doris / Postgres / S3 / Kafka...
        print(f"LOAD rows={payload['rows']} data={payload['data']}")

    load(transform(extract()))

    # e = extract()
    # t = transform(e)
    # l = load(t)

    # e >> t >> l

---

2. Airflow 平台核心能力

A. 可擴展的任務執行架構

• 功能：Airflow 採用「排程控制」與「任務執行」分離的分散式架構設計，任務可以分散執行於多個 Worker 節點，並依需求彈性擴充。
• 實際價值：
  • 資料量波動大、尖峰不固定 的工作負載
  • 可與容器平台（如 Kubernetes）整合，避免資源長期閒置
  • 不同類型的任務（ETL、API、ML）可以並行處理

B. 流程即程式碼的版本管理能力

• 功能：Airflow 將所有流程定義為程式碼，天然可納入 Git 等版本控制系統進行管理。
• 實際價值：
  • 每一次流程變更都有明確紀錄，方便回溯與審核
  • 可安全地重跑歷史資料（Backfill），確保使用的是「當時的流程邏輯」
  • 流程修改可以走標準的開發流程（Review / 測試 / 部署）

C. 多元觸發機制：時間 + 事件

• 功能：除了傳統的時間排程（Cron），Airflow 也支援以「資料狀態」或「事件」作為流程啟動條件。
• 實際價值：
  • 上游資料完成後，下游流程可立即啟動
  • 減少不必要的空跑（資料還沒好就先排程）
  • 更適合串接資料湖、串流處理或跨系統流程

D. 清楚可視化的流程監控與錯誤處理

• 功能：Airflow 提供 DAG 視覺化介面，清楚呈現任務依賴、執行狀態與錯誤位置。
• 實際價值：
  • 問題發生時，可以快速定位「卡在哪一個步驟」
  • 支援重試、跳過、補跑等操作，降低人工介入成本
  • 可與告警系統整合，提升維運可視性

---

3. 常見的數據流程管理痛點

傳統痛點	Airflow 解決方案
GUI 黑箱作業 流程邏輯藏在圖形介面與設定檔深處，版控困難，難以 Code Review。	Configuration as Code 流程即代碼。邏輯透明、可版控、可測試、可多人協作開發。
**依賴地獄 (Dependency Hell)** 任務 A 失敗，任務 B 卻繼續跑；或跨系統依賴難以管理。	DAG 狀態管理 嚴格的依賴控制，支援複雜的邏輯判斷 (Branching) 與跨 DAG 觸發。
授權費昂貴 商用軟體以 Agent 或 Job 數量計費，擴充成本極高。	Open Source 開源免費，無 Agent 數量限制，適合大規模雲端動態擴展。
人才斷層 年輕工程師不想學專有的商用工具指令。	Python 標準 使用通用的 Python 語言，人才庫龐大且易於招募。

---

4. Airflow vs. Control-M

比較維度	Apache Airflow 3.0	BMC Control-M
核心哲學	**Code-First (代碼優先)** 適合開發者 (Dev) 與資料工程師。	**Config-First (設定優先)** 適合維運人員 (Ops)，以 GUI 拖拉為主。
版控與協作	🏆 Git Flow 整合 天生支援 Branch、PR、Code Review 流程。	弱 依賴匯出 XML/JSON 進行版控，難以多人同時開發。
雲端與生態	🏆 **雲原生 (Cloud Native)** AWS/GCP/Azure 整合極深，容器化支援佳。	傳統強項 強項在 Mainframe (大型主機) 與地端 Legacy 系統。
成本結構	硬體與人力成本 軟體免費，但需投入工程人力維護。	高昂授權費 按 Job 數或處理器核心計費，擴充昂貴。
客製化能力	無限 任何 Python 能寫的邏輯都能跑。	受限 需依賴原廠提供的 Plugin 或自行開發複雜腳本。

兩者在實務上常並存，依工作負載性質選擇合適工具。

---

5. 總結

Airflow 能夠實現：

1. 自動化：將手動的腳本轉化為自動執行的流程。
2. 標準化：用統一的 Python 語法定義所有資料處理邏輯。
3. 視覺化：讓複雜的資料依賴關係變得清晰可見。
4. 可靠性：自動重試 (Retry)、超時控制 (Timeout) 與 警報通知 (Alerting)。