需求單位想要隨時自己調整一個參數，不用每次都找工程師
開資料庫 table 給他們改不好維護，存在 Airflow Variable 又沒辦法只開放一個
不想開太大的權限，也不想隨時待命幫忙改值，怎麼辦？

既有問題

有一支照排程持續執行的 DAG，裡面的判斷邏輯依賴一個由需求單位決定的參數
平常是固定值，但遇到特殊狀況就得臨時調整，而且要「改完，下一次執行就生效」

方案一：手動帶參數觸發 DAG

這支 DAG 一次 run 只處理 500 筆，總量可能有 6000 筆，要靠排程連續跑好幾輪才會處理完
參數必須在這幾輪排程裡都持續生效，靠每次手動帶參數觸發根本不實際

那先把 DAG 暫停，改成手動觸發呢？

DAG 暫停時手動觸發的 run 只會排進佇列，要等 DAG 重新開啟才會開始執行
而重新開啟的當下，只要已經錯過排程時間，scheduler 就會馬上補建排程 run

• catchup=False：補建最近一個區間的 run
• catchup=True：錯過的區間全部補跑

不管哪一種，補建的都是排程 run，用的是預設參數，不是我們手動輸入的值

手動 run 還沒跑，預設值的 run 已經先處理掉一批資料了

結果就是部分資料用錯誤的參數處理，所以這個方案直接出局

方案二：開一張 table 存這個參數

需求單位平常就習慣用 SQL 查詢、整理資料，所以他們的提案是開一張 table 專門存這個參數
要調整時自己下一句 UPDATE 就好，不用每次都找工程師改程式

但為了一個值開一張表，然後權限、schema、資料庫維護全都要工程師扛
而且還有一連串問題要先回答：

• 這張 table 是專門給這個參數用的嗎？那其他專案或需求也有參數要存，是不是每個都要再開一張？
• 如果其他人也能存，很快就會變成什麼都往裡丟的雜物抽屜，沒人說得清楚哪筆是誰的、還有沒有在用
• 直接 UPDATE 沒有任何驗證，值打錯了要等排程出錯才會發現
• 誰在什麼時候改了什麼，沒有留下紀錄

方案三：Airflow Variable

Airflow 本來就有 Variable 可以存這種設定值，DAG 每次執行時讀取，改完下一次執行就生效
看起來是最適合的方案

問題是 Airflow 的權限以「資源」為單位控管，Variables 的編輯權限一開就是全部
其他流程用到的 API token、帳號密碼也都存在 Variable 裡

只想讓需求單位改一個參數，結果連 token 跟密碼都一起交出去了

基於最小權限原則(PoLP)，這個權限不能開給需求單位
但也不想每次需要調整時，都要有工程師隨時待命幫忙改

解決方案

用 DAG 當 Variable 的專屬窗口

既然 Variable 的權限切不細，那就把「能改哪個 key」寫死在 DAG 裡

• 另外寫一支 DAG，寫死要更新的 Variable Key，需求單位只能透過 Trigger DAG 輸入新的值
• 透過 role 設定這支 DAG 的權限，只讓需求單位看得到、觸發得了
• 原本排程中的 DAG 完全不用動，照常讀取 Variable，下一次執行就會套用新值
• 每次修改都是一次手動 DAG run，誰在什麼時候改成什麼值都會留下紀錄

等於是把「改 Variable」包成一個只有一個欄位的櫃台
需求單位只能填值，不能選要改哪一個

概念驗證

from __future__ import annotations

import logging
from datetime import datetime, timedelta

from airflow.sdk import DAG, Param, Variable, task
from airflow.sdk.definitions.param import ParamsDict
from airflow.sdk.exceptions import AirflowException

logger = logging.getLogger(__name__)

VARIABLE_KEY = "my_variable"

default_args = {
    "owner": "Developer",
    "depends_on_past": False,
    "start_date": datetime(2026, 6, 14),
    "retry_delay": timedelta(minutes=1),
}


with DAG(
    "variable_setter",
    default_args=default_args,
    schedule=None,
    dag_display_name="Variable Setter",
    max_active_runs=1,
    catchup=False,
    params=ParamsDict(
        {
            "variable_value": Param(
                type="string",
                title="Variable Value",
                description="要設定的 Variable 值",
            ),
        }
    ),
    doc_md="""\
        ### 指定 Variable 值
        輸入 Variable 值，寫入指定 Variable Key。
    """,
) as dag:

    @task
    def update_variable(**kwargs):
        """取得傳入參數，驗證合法性後更新指定 Variable"""

        # 取得傳入參數，並判斷合法性
        variable_value = (kwargs["params"].get("variable_value") or "").strip()
        if not variable_value:
            raise AirflowException("variable_value must not be empty")

        # 取得原本的值
        old_value = Variable.get(VARIABLE_KEY, default=None)

        # 更新 Variable 的值
        Variable.set(VARIABLE_KEY, variable_value, description="還可以寫註解")
        logger.info(
            "Updated Variable %s: %s -> %s",
            VARIABLE_KEY,
            old_value,
            variable_value,
        )

        return variable_value

    # 驗證參數之後更新 Variable
    update_variable()

幾個重點：

• params 搭配 Param 定義輸入欄位，觸發 DAG 時會自動產生表單
• 寫入前先 strip() 並檢查空值，避免一個不小心把參數清空
• 寫入前先讀出舊值，log 記下 舊值 -> 新值，日後追查比較方便
• max_active_runs=1 避免兩個人同時觸發互相覆蓋
• Variable.set() 可以順便帶 description，在 Variable 列表裡就看得到用途

1. 初始狀態

一開始 my_variable 還不存在

2. 觸發 DAG 寫入值

觸發 DAG 並輸入 init_var，從 log 可以看到 my_variable: None -> init_var

回到 Variable 列表，值和描述都寫進去了

3. 再次觸發更新值

這次輸入 update_var，log 記錄了 init_var -> update_var

查看實際的 Variable 值，可以看到已經更新了

注意事項

要記得設定 role 權限，只給這支 DAG 的讀取／觸發權限
不要給 Variables 的編輯權限，否則等於白做

有固定格式的值一定要驗證
可以在 Param 加上 pattern，或在 task 裡檢查，避免輸入錯誤直接讓排程中的 DAG 出錯

讀取 Variable 的那一端要在 task 裡呼叫 Variable.get()，不要寫在 DAG 檔案最外層
否則每次 DAG parse 都會去查 metadata DB