本文將介紹 Airflow 這一款優(yōu)秀的調(diào)度工具沐兰。主要包括 Airflow 的服務(wù)構(gòu)成梧喷、Airflow 的 Web 界面、DAG 配置拉背、常用配置以及 Airflow DAG Creation Manager Plugin 這一款 Airflow 插件师崎。

一、什么是 Airflow

Airflow 是 Airbnb 開源的一個(gè)用 Python 編寫的調(diào)度工具椅棺。于 2014 年啟動(dòng)犁罩，2015 年春季開源，2016 年加入 Apache 軟件基金會(huì)的孵化計(jì)劃两疚。

Airflow 通過(guò) DAG 也即是有向非循環(huán)圖來(lái)定義整個(gè)工作流床估，因而具有非常強(qiáng)大的表達(dá)能力。

Airflow DAG

如上圖所示诱渤，一個(gè)工作流可以用一個(gè) DAG 來(lái)表示顷窒，在 DAG 中將完整得記錄整個(gè)工作流中每個(gè)作業(yè)之間的依賴關(guān)系、條件分支等內(nèi)容，并可以記錄運(yùn)行狀態(tài)鞋吉。通過(guò) DAG鸦做，我們可以精準(zhǔn)的得到各個(gè)作業(yè)之間的依賴關(guān)系。

在進(jìn)一步介紹 Airflow 之前谓着，我想先介紹一些在 Airflow 中常見(jiàn)的名詞概念：

• DAG

  DAG 意為有向無(wú)循環(huán)圖泼诱，在 Airflow 中則定義了整個(gè)完整的作業(yè)。同一個(gè) DAG 中的所有 Task 擁有相同的調(diào)度時(shí)間赊锚。

• Task

  Task 為 DAG 中具體的作業(yè)任務(wù)治筒，它必須存在于某一個(gè) DAG 之中。Task 在 DAG 中配置依賴關(guān)系舷蒲，跨 DAG 的依賴是可行的耸袜，但是并不推薦∩剑跨 DAG 依賴會(huì)導(dǎo)致 DAG 圖的直觀性降低堤框，并給依賴管理帶來(lái)麻煩。

• DAG Run

  當(dāng)一個(gè) DAG 滿足它的調(diào)度時(shí)間纵柿，或者被外部觸發(fā)時(shí)蜈抓，就會(huì)產(chǎn)生一個(gè) DAG Run“喝澹可以理解為由 DAG 實(shí)例化的實(shí)例沟使。

• Task Instance

  當(dāng)一個(gè) Task 被調(diào)度啟動(dòng)時(shí)，就會(huì)產(chǎn)生一個(gè) Task Instance渊跋±拔耍可以理解為由 Task 實(shí)例化的實(shí)例。

二拾酝、Airflow 的服務(wù)構(gòu)成

一個(gè)正常運(yùn)行的 Airflow 系統(tǒng)一般由以下幾個(gè)服務(wù)構(gòu)成

• WebServer

  Airflow 提供了一個(gè)可視化的 Web 界面燕少。啟動(dòng) WebServer 后，就可以在 Web 界面上查看定義好的 DAG 并監(jiān)控及改變運(yùn)行狀況微宝。也可以在 Web 界面中對(duì)一些變量進(jìn)行配置棺亭。

• Worker

  一般來(lái)說(shuō)我們用 Celery Worker 來(lái)執(zhí)行具體的作業(yè)。Worker 可以部署在多臺(tái)機(jī)器上蟋软，并可以分別設(shè)置接收的隊(duì)列镶摘。當(dāng)接收的隊(duì)列中有作業(yè)任務(wù)時(shí)，Worker 就會(huì)接收這個(gè)作業(yè)任務(wù)岳守，并開始執(zhí)行凄敢。Airflow 會(huì)自動(dòng)在每個(gè)部署 Worker 的機(jī)器上同時(shí)部署一個(gè) Serve Logs 服務(wù)，這樣我們就可以在 Web 界面上方便的瀏覽分散在不同機(jī)器上的作業(yè)日志了湿痢。

• Scheduler

  整個(gè) Airflow 的調(diào)度由 Scheduler 負(fù)責(zé)發(fā)起涝缝，每隔一段時(shí)間 Scheduler 就會(huì)檢查所有定義完成的 DAG 和定義在其中的作業(yè)扑庞，如果有符合運(yùn)行條件的作業(yè)，Scheduler 就會(huì)發(fā)起相應(yīng)的作業(yè)任務(wù)以供 Worker 接收拒逮。

• Flower

  Flower 提供了一個(gè)可視化界面以監(jiān)控所有 Celery Worker 的運(yùn)行狀況罐氨。這個(gè)服務(wù)并不是必要的。

三滩援、Airflow 的 Web 界面

下面簡(jiǎn)單介紹一下 Airflow 的 Web 操作界面栅隐，從而可以對(duì) Airflow 有一個(gè)更直觀的了解。

1玩徊、DAG 列表

DAG 列表

1. 左側(cè) On/Off 按鈕控制 DAG 的運(yùn)行狀態(tài)租悄，Off 為暫停狀態(tài)，On 為運(yùn)行狀態(tài)恩袱。注意：所有 DAG 腳本初次部署完成時(shí)均為 Off 狀態(tài)泣棋。

2. 若 DAG 名稱處于不可點(diǎn)擊狀態(tài)，可能為 DAG 被刪除或未載入畔塔。若 DAG 未載入潭辈，可點(diǎn)擊右側(cè)刷新按鈕進(jìn)行刷新。注意：由于可以部署若干 WebServer俩檬，所以單次刷新可能無(wú)法刷新所有 WebServer 緩存萎胰，可以嘗試多次刷新碾盟。

3. Recent Tasks 會(huì)顯示最近一次 DAG Run（可以理解為 DAG 的執(zhí)行記錄）中 Task Instances（可以理解為作業(yè)的執(zhí)行記錄）的運(yùn)行狀態(tài)棚辽，如果 DAG Run 的狀態(tài)為 running，此時(shí)顯示最近完成的一次以及正在運(yùn)行的 DAG Run 中所有 Task Instances 的狀態(tài)冰肴。

4. Last Run 顯示最近一次的 execution date屈藐。注意：execution date 并不是真實(shí)執(zhí)行時(shí)間，具體細(xì)節(jié)在下文 DAG 配置中詳述熙尉。將鼠標(biāo)移至 execution date 右側(cè) info 標(biāo)記上联逻，會(huì)顯示 start date，start date 為真實(shí)運(yùn)行時(shí)間检痰。start date 一般為 execution date 所對(duì)應(yīng)的下次執(zhí)行時(shí)間包归。

2、作業(yè)操作框

在 DAG 的樹狀圖和 DAG 圖中都可以點(diǎn)擊對(duì)應(yīng)的 Task Instance 以彈出 Task Instance 模態(tài)框铅歼，以進(jìn)行 Task Instance 的相關(guān)操作公壤。注意：選擇的 Task Instance 為對(duì)應(yīng) DAG Run 中的 Task Instance。

作業(yè)操作框

1. 在作業(yè)名字的右邊有一個(gè)漏斗符號(hào)椎椰，點(diǎn)擊后整個(gè) DAG 的界面將只顯示該作業(yè)及該作業(yè)的依賴作業(yè)厦幅。當(dāng)該作業(yè)所處的 DAG 較大時(shí)，此功能有較大的幫助慨飘。

2. Task Instance Details 顯示該 Task Instance 的詳情确憨，可以從中得知該 Task Instance 的當(dāng)前狀態(tài)，以及處于當(dāng)前狀態(tài)的原因。例如休弃，若該 Task Instance 為 no status 狀態(tài)吞歼，遲遲不進(jìn)入 queued 及 running 狀態(tài)，此時(shí)就可通過(guò) Task Instance Details 中的 Dependency 及 Reason 得知原因塔猾。

3. Rendered 顯示該 Task Instance 被渲染后的命令浆熔。

4. Run 指令可以直接執(zhí)行當(dāng)前作業(yè)。

5. Clear 指令為清除當(dāng)前 Task Instance 狀態(tài)桥帆，清除任意一個(gè) Task Instance 都會(huì)使當(dāng)前 DAG Run 的狀態(tài)變更為 running医增。注意：如果被清除的 Task Instance 的狀態(tài)為 running，則會(huì)嘗試 kill 該 Task Instance 所執(zhí)行指令老虫，并進(jìn)入 shutdown 狀態(tài)叶骨，并在 kill 完成后將此次執(zhí)行標(biāo)記為 failed（如果 retry 次數(shù)沒(méi)有用完，將標(biāo)記為 up_for_retry）祈匙。Clear 有額外的5個(gè)選項(xiàng)忽刽，均為多選，這些選項(xiàng)從左到右依次為：

   • Past: 同時(shí)清除所有過(guò)去的 DAG Run 中此 Task Instance 所對(duì)應(yīng)的 Task Instance夺欲。
   • Future: 同時(shí)清除所有未來(lái)的 DAG Run 中此 Task Instance 所對(duì)應(yīng)的 Task Instance跪帝。注意：僅清除已生成的 DAG Run 中的 Task Instance。
   • Upstream: 同時(shí)清除該 DAG Run 中所有此 Task Instance 上游的 Task Instance些阅。
   • Downstream: 同時(shí)清除該 DAG Run 中所有此 Task Instance 下游的 Task Instance伞剑。
   • Recursive: 當(dāng)此 Task Instance 為 sub DAG 時(shí)，循環(huán)清除所有該 sub DAG 中的 Task Instance市埋。注意：若當(dāng)此 Task Instance 不是 sub DAG 則忽略此選項(xiàng)黎泣。

6. Mark Success 指令為講當(dāng)前 Task Instance 狀態(tài)標(biāo)記為 success。注意：如果該 Task Instance 的狀態(tài)為 running缤谎，則會(huì)嘗試 kill 該 Task Instance 所執(zhí)行指令抒倚，并進(jìn)入 shutdown 狀態(tài)，并在 kill 完成后將此次執(zhí)行標(biāo)記為 failed（如果 retry 次數(shù)沒(méi)有用完坷澡，將標(biāo)記為 up_for_retry）托呕。

四、DAG 配置

Airflow 中的 DAG 是由 Python 腳本來(lái)配置的频敛，因而可擴(kuò)展性非常強(qiáng)项郊。Airflow 提供了一些 DAG 例子，我們可以通過(guò)一個(gè)例子來(lái)簡(jiǎn)單得了解一下姻政。

# -*- coding: utf-8 -*-

import airflow
from airflow.operators.bash_operator import BashOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.models import DAG


args = {
    'owner': 'airflow',
    'start_date': airflow.utils.dates.days_ago(2)
}

dag = DAG(
    dag_id='example_bash_operator', default_args=args,
    schedule_interval='0 0 * * *')

cmd = 'ls -l'
run_this_last = DummyOperator(task_id='run_this_last', dag=dag)

run_this = BashOperator(
    task_id='run_after_loop', bash_command='echo 1', dag=dag)
run_this.set_downstream(run_this_last)

for i in range(3):
    i = str(i)
    task = BashOperator(
        task_id='runme_'+i,
        bash_command='echo "{{ task_instance_key_str }}" && sleep 1',
        dag=dag)
    task.set_downstream(run_this)

task = BashOperator(
    task_id='also_run_this',
    bash_command='echo "run_id={{ run_id }} | dag_run={{ dag_run }}"',
    dag=dag)
task.set_downstream(run_this_last)

我們可以看到呆抑，整個(gè) DAG 的配置就是一份完整的 Python 代碼，在代碼中實(shí)例化 DAG汁展，實(shí)例化適合的 Operator鹊碍，并通過(guò) set_downstream 等方法配置上下游依賴關(guān)系厌殉。下面我們簡(jiǎn)單看一下在 DAG 配置中的幾個(gè)重要概念。

• DAG

  要配置一個(gè) DAG 自然需要一個(gè) DAG 實(shí)例侈咕。在同一個(gè) DAG 下的所有作業(yè)公罕，都需要將它的 dag 屬性設(shè)置為這個(gè) DAG 實(shí)例。在實(shí)例化 DAG 時(shí)耀销，通過(guò)傳參數(shù)可以給這個(gè) DAG 實(shí)例做一些必要的配置楼眷。

  • dag_id

    給 DAG 取一個(gè)名字，方便日后維護(hù)熊尉。

  • default_args

    默認(rèn)參數(shù)罐柳，當(dāng)屬于這個(gè) DAG 實(shí)例的作業(yè)沒(méi)有配置相應(yīng)參數(shù)時(shí)，將使用 DAG 實(shí)例的 default_args 中的相應(yīng)參數(shù)狰住。

  • schedule_interval

    配置 DAG 的執(zhí)行周期张吉，語(yǔ)法和 crontab 的一致。

• 作業(yè) (Task)

  Airflow 提供了很多 Operator催植，我們也可以自行編寫新的 Operator肮蛹。在本例中使用了 2 種 Operator，DummyOperator 什么都不會(huì)做创南， BashOperator 則會(huì)執(zhí)行 bash_command 參數(shù)所指定的 bash 指令伦忠，并且使用 jinja2 模版引擎，對(duì)該指令進(jìn)行渲染稿辙，因而在本例的 bash_command 中昆码，可以看到一些需要渲染的變量。當(dāng) Operator 被實(shí)例化后邓深，我們稱之為相應(yīng) DAG 的一個(gè)作業(yè)（Task）未桥。在實(shí)例化 Operator 時(shí)笔刹，同樣可以通過(guò)穿參數(shù)進(jìn)行必要的配置芥备，值得注意的是，如果在 DAG 中有設(shè)置 default_args 而在 Operator 中沒(méi)有覆蓋相應(yīng)配置舌菜，則會(huì)使用 default_args 中的配置萌壳。

  • dag

    傳遞一個(gè) DAG 實(shí)例，以使當(dāng)前作業(yè)屬于相應(yīng) DAG日月。

  • task_id

    給作業(yè)去一個(gè)名字袱瓮，方便日后維護(hù)。

  • owner

    作業(yè)的擁有者爱咬，方便作業(yè)維護(hù)尺借。另外有些 Operator 會(huì)根據(jù)該參數(shù)實(shí)現(xiàn)相應(yīng)的權(quán)限控制。

  • start_date

    作業(yè)的開始時(shí)間精拟，即作業(yè)將在這個(gè)時(shí)間點(diǎn)以后開始調(diào)度燎斩。

• 依賴

  配置以來(lái)的方法有兩種虱歪，除了可以使用作業(yè)實(shí)例的 set_upstream 和 set_downstream 方法外，還可以使用類似

  task1 << task2 << task3
  task3 >> task4
  

  這樣更直觀的語(yǔ)法來(lái)設(shè)置栅表。

這里我們要特別注意一個(gè)關(guān)于調(diào)度執(zhí)行時(shí)間的問(wèn)題笋鄙。在談這個(gè)問(wèn)題前，我們先確定幾個(gè)名詞：

• start date: 在配置中怪瓶，它是作業(yè)開始調(diào)度時(shí)間萧落。而在談?wù)搱?zhí)行狀況時(shí)，它是調(diào)度開始時(shí)間洗贰。
• schedule interval: 調(diào)度執(zhí)行周期找岖。
• execution date: 執(zhí)行時(shí)間，在 Airflow 中稱之為執(zhí)行時(shí)間敛滋，但其實(shí)它并不是真實(shí)的執(zhí)行時(shí)間宣增。

那么現(xiàn)在，讓我們看一下當(dāng)一個(gè)新配置的 DAG 生效后第一次調(diào)度會(huì)在什么時(shí)候矛缨。很多人會(huì)很自然的認(rèn)為爹脾，第一次的調(diào)度時(shí)間當(dāng)然是在作業(yè)中配置的 start date，但其實(shí)并不是箕昭。第一次調(diào)度時(shí)間是在作業(yè)中配置的 start date 的第二個(gè)滿足 schedule interval 的時(shí)間點(diǎn)灵妨，并且記錄的 execution date 為作業(yè)中配置的 start date 的第一個(gè)滿足 schedule interval 的時(shí)間點(diǎn)。聽(tīng)起來(lái)很繞落竹，讓我們來(lái)舉個(gè)例子泌霍。

假設(shè)我們配置了一個(gè)作業(yè)的 start date 為 2017年10月1日，配置的 schedule interval 為 **00 12 * * *** 那么第一次執(zhí)行的時(shí)間將是 2017年10月2日 12點(diǎn) 而此時(shí)記錄的 execution date 為 2017年10月1日 12點(diǎn)述召。因此 execution date 并不是如其字面說(shuō)的表示執(zhí)行時(shí)間朱转，真正的執(zhí)行時(shí)間是 execution date 所顯示的時(shí)間的下一個(gè)滿足 schedule interval 的時(shí)間點(diǎn)。

另外积暖，當(dāng)作業(yè)已經(jīng)執(zhí)行過(guò)之后藤为，start date 的配置將不會(huì)再生效，這個(gè)作業(yè)的調(diào)度開始時(shí)間將直接按照上次調(diào)度所對(duì)應(yīng)的 execution date 來(lái)計(jì)算夺刑。

這個(gè)例子只是簡(jiǎn)要的介紹了一下 DAG 的配置缅疟，也只介紹了非常少量的配置參數(shù)。Airflow 為 DAG 和作業(yè)提供了大量的可配置參數(shù)遍愿，詳情可以參考 Airflow 官方文檔存淫。

五、常用配置

在日常工作中沼填，有時(shí)候僅僅靠配置作業(yè)依賴和調(diào)度執(zhí)行周期并不能滿足一些復(fù)雜的需求桅咆。接下來(lái)將介紹一些常用的作業(yè)配置。

1坞笙、跳過(guò)非最新 DAG Run

假如有一個(gè)每小時(shí)調(diào)度的 DAG 出錯(cuò)了岩饼，我們把它的調(diào)度暫停刽脖，之后花了3個(gè)小時(shí)修復(fù)了它，修復(fù)完成后重新啟動(dòng)這個(gè)作業(yè)的調(diào)度忌愚。于是 Airflow 一下子創(chuàng)建了 3 個(gè) DAG Run 并同時(shí)執(zhí)行曲管，這顯然不是我們希望的，我們希望它只執(zhí)行最新的 DAG Run硕糊。

我們可以創(chuàng)建一個(gè) Short Circuit Operator院水，并且讓 DAG 中所有沒(méi)有依賴的作業(yè)都依賴這個(gè)作業(yè)，然后在這個(gè)作業(yè)中進(jìn)行判斷简十，檢測(cè)當(dāng)前 DAG Run 是否為最新檬某，不是最新的直接跳過(guò)整個(gè) DAG。

def skip_dag_not_latest_worker(ds, **context):
    if context['dag_run'] and context['dag_run'].external_trigger:
        logging.info('Externally triggered DAG_Run: allowing execution to proceed.')
        return True

    skip = False
    now = datetime.now()
    left_window = context['dag'].following_schedule(context['execution_date'])
    right_window = context['dag'].following_schedule(left_window)
    logging.info('Checking latest only with left_window: %s right_window: %s now: %s', left_window, right_window, now)

    if not left_window < now <= right_window:
        skip = True
    return not skip

ShortCircuitOperator(
    task_id='skip_dag_not_latest',
    provide_context=True,
    python_callable=skip_dag_not_latest_worker,
    dag=dag
)

2螟蝙、當(dāng)存在正在執(zhí)行的 DAG Run 時(shí)跳過(guò)當(dāng)前 DAG Run

依舊是之前提到的每小時(shí)調(diào)度的 DAG恢恼，假設(shè)它這次沒(méi)有出錯(cuò)而是由于資源、網(wǎng)絡(luò)或者其他問(wèn)題導(dǎo)致執(zhí)行時(shí)間變長(zhǎng)胰默，當(dāng)下一個(gè)調(diào)度時(shí)間開始時(shí) Airflow 依舊會(huì)啟動(dòng)一次新的 DAG Run场斑，這樣就會(huì)同時(shí)出現(xiàn) 2 個(gè) DAG Run。如果我們想要避免這種情況牵署，一個(gè)簡(jiǎn)單的方法是直接將 DAG 的 max_active_runs 設(shè)置為 1漏隐。但這樣會(huì)導(dǎo)致 DAG Run 堆積的問(wèn)題，如果你配置的調(diào)度是早上 9 點(diǎn)至晚上 9 點(diǎn)奴迅，直至晚上 9 點(diǎn)之后 Airflow 可能依舊在處理堆積的 DAG Run青责。這樣就可能影響到我們?cè)景才旁谕砩?9 點(diǎn)之后的任務(wù)。

我們可以創(chuàng)建一個(gè) Short Circuit Operator取具，并且讓 DAG 中所有沒(méi)有依賴的作業(yè)都依賴這個(gè)作業(yè)脖隶，然后在這個(gè)作業(yè)中進(jìn)行判斷，檢測(cè)當(dāng)前是否存在正在執(zhí)行的 DAG Run暇检，存在時(shí)則直接跳過(guò)整個(gè) DAG产阱。

def skip_dag_when_previous_running_worker(ds, **context):
    if context['dag_run'] and context['dag_run'].external_trigger:
        logging.info('Externally triggered DAG_Run: allowing execution to proceed.')
        return True

    skip = False
    session = settings.Session()
    count = session.query(DagRun).filter(
        DagRun.dag_id == context['dag'].dag_id,
        DagRun.state.in_(['running']),
    ).count()
    session.close()
    logging.info('Checking running DAG count: %s' % count)
    skip = count > 1
    return not skip

ShortCircuitOperator(
    task_id='skip_dag_when_previous_running',
    provide_context=True,
    python_callable=skip_dag_when_previous_running_worker,
    dag=dag
)

3、Sensor 的替代方案

Airflow 中有一類 Operator 被稱為 Sensor占哟，Sensor 可以感應(yīng)預(yù)先設(shè)定的條件是否滿足（如：某個(gè)時(shí)間點(diǎn)是否達(dá)到心墅、某條 MySQL 記錄是否被更新、某個(gè) DAG 是否完成）榨乎，當(dāng)滿足條件后 Sensor 作業(yè)變?yōu)?Success 使得下游的作業(yè)能夠執(zhí)行。Sensor 的功能很強(qiáng)大但卻帶來(lái)一個(gè)問(wèn)題瘫筐，假如我們有一個(gè) Sensor 用于檢測(cè)某個(gè) MySQL 記錄是否被更新蜜暑，在 Sensor 作業(yè)啟動(dòng)后 3 個(gè)小時(shí)這個(gè) MySQL 記錄才被更新。于是我們的這個(gè) Sensor 占用了一個(gè) Worker 整整 3 小時(shí)策肝，這顯然是一個(gè)極大的浪費(fèi)肛捍。

因此我們需要一個(gè) Sensor 的替代方案隐绵，既能滿足 Sensor 原來(lái)的功能，又能節(jié)省 Worker 資源拙毫。有一個(gè)辦法是不使用 Sensor依许，直接使用 Python Operator 判斷預(yù)先設(shè)定的條件是否滿足，如果不滿足直接 raise Exception缀蹄，然后將這個(gè)作業(yè)的 retry_delay（重試間隔時(shí)間） 設(shè)為每次檢測(cè)的間隔時(shí)間峭跳，retries（重試次數(shù)） 設(shè)為最長(zhǎng)檢測(cè)時(shí)間除以 retry_delay，即滿足：最長(zhǎng)檢測(cè)時(shí)間 = retries * retry_delay缺前。這樣既不會(huì)長(zhǎng)時(shí)間占用 Worker 資源蛀醉，又可以滿足 Sensor 原來(lái)的功能。

六衅码、Airflow DAG Creation Manager Plugin

正如上兩章所描述的拯刁，Airflow 雖然具有強(qiáng)大的功能，但是配置 DAG 并不是簡(jiǎn)單的工作逝段，也有一些較為繁瑣的概念垛玻，對(duì)于業(yè)務(wù)人員來(lái)說(shuō)可能略顯復(fù)雜。因此奶躯，筆者編寫了 Airflow DAG Creation Manager Plugin（https://github.com/lattebank/airflow-dag-creation-manager-plugin）以提供一個(gè) Web界面來(lái)讓業(yè)務(wù)人員可視化的編寫及管理 DAG夭谤。具體的安裝及使用方法請(qǐng)查看插件的README。

DAG 管理插件

如上圖所示巫糙，插件的 Web 界面中可以直接所見(jiàn)即所得的編寫 DAG 圖朗儒。

插件中盡量簡(jiǎn)化了一些繁瑣的諸如上文所述的作業(yè)開始調(diào)度時(shí)間等一系列的概念，并提供了一些在實(shí)際工作中常常會(huì)用到的一些額外的功能（如上文提到的跳過(guò)非最新 DAG Run参淹、當(dāng)存在正在執(zhí)行的 DAG Run 時(shí)跳過(guò)當(dāng)前 DAG Run 等）醉锄，以及版本控制和權(quán)限管理。如果大家在使用 Airflow 的過(guò)程中也有類似的問(wèn)題浙值，歡迎嘗試使用 Airflow DAG Creation Manager Plugin恳不。

七、總結(jié)

Airflow 適用于調(diào)度作業(yè)較為復(fù)雜开呐，特別是各作業(yè)之間的依賴關(guān)系復(fù)雜的情況烟勋。

希望本文能讓大家對(duì) Airflow 有所了解，并能將 Airflow 運(yùn)用到適合它使用的場(chǎng)景中筐付。

更多閱讀：
基于Redis的限流系統(tǒng)的設(shè)計(jì)
Presto+Hive統(tǒng)一賬戶體系及查詢監(jiān)控輕型解決方案
談?wù)劵?OpenResty 的接口網(wǎng)關(guān)設(shè)計(jì)

“泛金融技術(shù)”微信公眾號(hào)