近日，調(diào)度系統(tǒng)遷移工具 Air2phin 宣布開源。借助 Air2phin，用戶可 2 步將調(diào)度系統(tǒng)從 Airflow 遷移至 Apache DolphinScheduler，為有調(diào)度系統(tǒng)遷移需要的用戶帶來極大便利。

Air2phin 是什么？

Air2phin 是一個(gè)最近宣布開源的調(diào)度系統(tǒng)遷移工具，旨在將 Apache Airflow DAGs 文件轉(zhuǎn)換成 Apache DolphinScheduler python SDK 定義文件，從而實(shí)現(xiàn)用戶將調(diào)度系統(tǒng)（Workflow orchestration）從 Airflow 遷移到 DolphinScheduler 的目的。它是一個(gè)基于多規(guī)則的 AST 轉(zhuǎn)換器，使用 LibCST 來解析和轉(zhuǎn)換 Airflow 的 DAG 代碼，其全部規(guī)則使用 Yaml 文件定義，并提供了一定的自定義規(guī)則擴(kuò)展能力。

近期，Air2phin 已經(jīng)發(fā)布了0.0.12 版本，提供了豐富的功能，可以更好地幫助用戶完成 Airflow 到 Apache DolphinScheduler 的遷移。

AST 是 Abstract Syntax Tree（抽象語法樹）的縮寫，它是一種以樹狀結(jié)構(gòu)表示代碼語法結(jié)構(gòu)的數(shù)據(jù)結(jié)構(gòu)。在編譯器中，AST 是由詞法分析器和語法分析器生成的。詞法分析器將源代碼轉(zhuǎn)換成標(biāo)記流（token stream），語法分析器將標(biāo)記流轉(zhuǎn)換成抽象語法樹。AST 是一種樹狀結(jié)構(gòu)，它由一系列節(jié)點(diǎn)組成，每個(gè)節(jié)點(diǎn)表示代碼中的一個(gè)語法結(jié)構(gòu)（如表達(dá)式、語句、函數(shù)、類等），節(jié)點(diǎn)之間的關(guān)系表示語法結(jié)構(gòu)之間的嵌套關(guān)系。

為什么開源 Air2phin？

可能有人會(huì)問，為什么我需要一個(gè)遷移工具？這是因?yàn)殡S著業(yè)務(wù)的發(fā)展，企業(yè)或組織原來使用的工作流編排系統(tǒng)已經(jīng)無法滿足當(dāng)前的需求，需要將工作流編排系統(tǒng)遷移到新的平臺(tái)或者更新到新的版本。經(jīng)過調(diào)研，很多用戶有了將調(diào)度系統(tǒng)從開源工作流編排系統(tǒng) Airflow 遷移到 Apache DolphinScheduler 上來的需求。

在遷移過程中，由于數(shù)據(jù)處理任務(wù)可能涉及多個(gè)系統(tǒng)之間的依賴關(guān)系，遷移過程需要確保在不影響業(yè)務(wù)運(yùn)行的前提下完成。此時(shí)，調(diào)度系統(tǒng)遷移工具就可以發(fā)揮重要作用，它能減少人工干預(yù)，盡量自動(dòng)化地完成兩個(gè)調(diào)度系統(tǒng)間的遷移工作，并且能兼容多個(gè)系統(tǒng)間的多個(gè)版本，幾乎可以做到用戶無干預(yù)完成遷移。

為此，白鯨開源專門研發(fā)了開源遷移工具 Air2phin，可以讓用戶 2 步將調(diào)度系統(tǒng)從 Airflow 遷移至 Apache DolphinScheduler，為用戶帶來極大的便利。

為了讓大家更好地理解 Air2phin 的重要性，我們先從調(diào)度系統(tǒng)的相關(guān)背景知識(shí)開始，了解將調(diào)度系統(tǒng)從 Airflow 遷移至 Apache DolphinScheduler 的好處。

為什么要從 Airflow 遷移至 DolphinScheduler？

什么是工作流編排系統(tǒng)？

工作流編排系統(tǒng)，是以尊重編排規(guī)則和業(yè)務(wù)邏輯的方式管理數(shù)據(jù)流。工作流編排工具讓用戶可以將多個(gè)有關(guān)聯(lián)的任務(wù)轉(zhuǎn)換為可以安排、運(yùn)行和觀測的工作流，幫助企業(yè)更好地管理和控制業(yè)務(wù)流程，從而提高業(yè)務(wù)效率。工作流編排是數(shù)據(jù)處理流程中不可或缺的組件之一，負(fù)責(zé)根據(jù)預(yù)先定義的規(guī)則和邏輯執(zhí)行數(shù)據(jù)處理任務(wù)，確保數(shù)據(jù)處理流程按照預(yù)期順利執(zhí)行，常見工作流編排系統(tǒng)包括 Apache DolphinScheduler、Apache Airflow、Apache Oozie, Azkaban 等。

Airflow是什么？

其中，Apache Airflow 是一個(gè)開源的工作流編排系統(tǒng)，它可以幫助用戶創(chuàng)建、調(diào)度和監(jiān)控復(fù)雜的工作流程。Airflow 最初由 Airbnb 開發(fā)，并于 2016 年開源，現(xiàn)在由 Apache 軟件基金會(huì)維護(hù)。Airflow 使用 Python 語言編寫，具有高度的可擴(kuò)展性和靈活性，支持多種任務(wù)類型，如計(jì)算、數(shù)據(jù)處理、通知、交互等。Airflow 的工作流程是通過編寫 Python 腳本來定義的，可以使用 Airflow 提供的操作符和鉤子，以及自定義操作符和鉤子來擴(kuò)展其功能。但其有著不可忽視的缺陷，比如需要需要深度二次開發(fā)，脫離社區(qū)版本，升級(jí)成本高；Python 技術(shù)棧維護(hù)迭代成本高；scheduler loop 掃描 Dag folder 延遲降低性能的問題；以及在生產(chǎn)環(huán)境中使用穩(wěn)定性差等。

在新數(shù)據(jù)時(shí)代業(yè)務(wù)需求下誕生的 Apache DolphinScheduler 是一個(gè)開源的分布式工作流調(diào)度系統(tǒng)，彌補(bǔ)了以往調(diào)度系統(tǒng)的弱勢(shì)，旨在為企業(yè)用戶提供一種可靠、高效、易于使用的工作流調(diào)度平臺(tái)，支持多種任務(wù)類型，如計(jì)算、數(shù)據(jù)處理、ETL 等。

與 Airflow 相比，DolphinScheduler 采用了分布式架構(gòu)，提供了多種任務(wù)類型，用戶可以定義任務(wù)之間的依賴關(guān)系，設(shè)置任務(wù)的優(yōu)先級(jí)和調(diào)度策略等，其使用可視化的界面來創(chuàng)建和管理工作流程的特性更是與 Airflow 形成鮮明對(duì)比，變得更加易于操作，對(duì)非編程人員來說更加友好。

經(jīng)過調(diào)研對(duì)比，對(duì)于很多用戶來說，將調(diào)度系統(tǒng)遷移至 Apache DolphinScheduler 是一個(gè)降本增效的更優(yōu)選擇。

Air2phin 如何安裝和使用

Air2phin 是一個(gè) python 的包，可以通過 Python 的包安裝工具 pip 完成安裝，詳見air2phin getting start。

python -m pip install --upgrade air2phin

一個(gè)簡單的例子

我們通過一個(gè)簡單的例子，來說明如何使用 Air2phin 的。我們截取了 airflow tutorial.py 中的部分代碼作為 Air2phin 轉(zhuǎn)化的例子，來說明 Air2phin 如何逐步完成轉(zhuǎn)化成 dolphinscheduler python sdk。

圖1：airflow tutorial.py 中的部分代碼

圖2：Air2phin 如何逐步完成轉(zhuǎn)化成 dolphinscheduler python sdk

假設(shè)將 airflow tutorial.py 部分內(nèi)容保存至文件 tutorial_part.py，想要將其轉(zhuǎn)化成 dolphinscheduler python sdk 定義，只需要一行命令就能完成。結(jié)果如圖 2 所示，因?yàn)槊钤黾恿?–inplace 參數(shù)，所以 Air2phin 會(huì)直接將原文件覆蓋，如果不需要覆蓋原問題，可以不使用 –inplace 參數(shù)，Air2phin 會(huì)新增一個(gè) tutorial_part-air2phin.py 文件來保存轉(zhuǎn)化后的內(nèi)容。

air2phin migrate --inplace tutorial_part.py

通過觀察，我們發(fā)現(xiàn)這次轉(zhuǎn)化分別觸發(fā)了多條轉(zhuǎn)化規(guī)則，包括

• 將 airflow.DAG 轉(zhuǎn)換成

  pydolphinscheduler.core.process_definition.ProcessDefinition，這個(gè)規(guī)則在第三行（import語句）以及第六行 DAG context

• 將 airflow.Operators.bash.BashOperator 轉(zhuǎn)換成

  pydolphinscheduler.tasks.shell.Shell，這個(gè)規(guī)則在任務(wù) t1，t2 中都被使用

• 除了對(duì)應(yīng)的類轉(zhuǎn)化之外，我們需要將類的屬性進(jìn)行轉(zhuǎn)化，如將

  airflow.DAG.schedule_interval 轉(zhuǎn)換成了 ProcessDefinition.schedule，同時(shí)修改了部分值的內(nèi)容，如將 timedelta(days=1) 轉(zhuǎn)成 ‘0 0 0 * * ? *’

最后，我們只需要安裝 pydolphinscheduler ，并且將轉(zhuǎn)化后的文件通過 python 運(yùn)行，就能完成工作流的遷移了，詳見 pydolphinscheduler 使用(https://dolphinscheduler.apache.org/python/main/start.html#installing-pydolphinscheduler)。

# 安裝 apache-dolphinschedulerpython -m pip install apache-dolphinscheduler# 將工作流提交到 dolphinschedulerpython tutorial_part.py

在運(yùn)行 python tutorial_part.py 時(shí)，需要保證 dolphinscheduler API 和 python gateway 服務(wù)已經(jīng)啟動(dòng)，并且開放了對(duì)應(yīng)的端口，詳見啟動(dòng) python gateway service。

至此，我們通過一個(gè)簡單的例子，說明了 Air2phin 是如何完成遷移的。

工作原理

Airflow 和 dolphinscheduler python sdk 如何工作？

在了解 Air2phin 如果工作之前，先了解 Airflow 和 dolphinscheduler python sdk 如何工作是非常重要的前置條件，幫助我們更好地了解 Air2phin 的遷移步驟，當(dāng)遇到問題的時(shí)候也能更加從容地應(yīng)對(duì)。

Airflow 如何工作：A

irflow 工作流相關(guān)的信息都保存在 DAG 文件中，之后將 DAG 文件放置到 Airflow 的指定目錄，Airflow 的 Scheduler 會(huì)間隔一定時(shí)間去掃描和解析 Airflow 的 DAG 文件，所以 DAG 文件是被動(dòng)被掃描和更新的。

dolphinscheduler python sdk:

同 Airflow 類似，將全部工作流相關(guān)的信息都通過 Python 文件定義，但是 dolphinscheduler python sdk 是通過人為主動(dòng)觸發(fā)的方式，將工作流信息提交，運(yùn)行命令 python 工作流文件名 即可完成主動(dòng)任務(wù)提交。

Air2phin 工作流程

了解完兩者是如何使用，如何提交/發(fā)現(xiàn)工作流的，將更加利于我們對(duì) Air2phin 的工作原理的理解。因?yàn)?Airflow 的 DAG 文件以及 DolphinScheduler 的 Python sdk 定義文件都是 Python 編寫的，所以 Air2phin 的大部分代碼都是處理兩者間的差異，最后將 Airflow 的代碼轉(zhuǎn)化成 dolphinscheduler python sdk 和定義。

Air2phin 使用了 LibCST(https://libcst.readthedocs.io/en/latest/) 來實(shí)現(xiàn) airflow python DAG 代碼的抽象語法樹解析，然后通過 LibCST 的 Transformer(https://libcst.readthedocs.io/en/latest/tutorial.html#Build-Visitor-or-Transformer)結(jié)合轉(zhuǎn)化規(guī)則最后轉(zhuǎn)化成 dolphinscheduler python sdk 的定義。

Air2phin 整體工作流程如下：

• 從標(biāo)準(zhǔn)輸入或者文件中獲取原本的 Airflow DAG 內(nèi)容

• 從 Yaml 文件加載所有轉(zhuǎn)換規(guī)則

• 將 Airflow DAG 內(nèi)容通過 LibCST 解析成 CST 樹

• 通過 LibCST Transformer 轉(zhuǎn)換 dolphinscheduler python sdk 定義內(nèi)容

Air2phin 最佳實(shí)踐

• 遷移整個(gè)文件夾而不是單個(gè)文件

當(dāng)用戶想要遷移 Airflow 到 DolphinScheduler 的時(shí)候，都是想要整體做遷移而不是單個(gè)文件遷移的，Air2phin 提供整體文件夾遷移的能力，只需要將路徑從文件路徑改成文件夾即可。

# 遷移整個(gè) ~/airflow/dags 文件夾air2phin migrate --inplace ~/airflow/dags

• 增加自定義的規(guī)則

部分使用 Airflow 的用戶自定義 Hook 或者 operator，用戶自定義的 Operator 無法通過 Air2phin 內(nèi)置的轉(zhuǎn)化規(guī)則完成轉(zhuǎn)化，需要用戶增加自定義的規(guī)則，并告訴 Air2phin 規(guī)則的位置。例如我們有一個(gè)叫 MyCustomOperator 的算子是繼承 PostgresOperator 的大部分功能， 只是命名不一樣，其定義如下：

from airflow.providers.postgres.operators.postgres import PostgresOperatorclass MyCustomOperator(PostgresOperator): def __init__( self, *, sql: str | Iterable[str], my_custom_conn_id: str = 'postgres_default', autocommit: bool = False, parameters: Iterable | Mapping | None = None, database: str | None = None, runtime_parameters: Mapping | None = None, **kwargs, ) -> None: super.__init__( sql=sql, postgres_conn_id=my_custom_conn_id, autocommit=autocommit, parameters=parameters, database=database, runtime_parameters=runtime_parameters, **kwargs, )

它在 Airflow 的多個(gè) DAG 中被使用，使用的方式如下：

from custom.my_custom_operator import MyCustomOperatorwith DAG( dag_id='my_custom_dag', default_args=default_args, schedule_interval='@once', start_date=days_ago(2), tags=['example'],) as dag: t1 = MyCustomOperator( task_id='my_custom_task', sql='select * from table', my_custom_conn_id='my_custom_conn_id', )

現(xiàn)在需要對(duì)這個(gè) Operator 進(jìn)行轉(zhuǎn)化，我們可以自定義一個(gè)轉(zhuǎn)化規(guī)則，并將其命名為 MyCustomOperator.yaml，內(nèi)容如下，最主要的內(nèi)容是 migration.module 和 migration.parameter 的定義，其確定了轉(zhuǎn)化規(guī)則：

name: MyCustomOperatordescription: The configuration for migrating airflow custom operator MyCustomOperator to DolphinScheduler SQL task.migration: module: - action: replace src: custom.my_custom_operator.MyCustomOperator dest: pydolphinscheduler.tasks.sql.Sql parameter: - action: replace src: task_id dest: name - action: replace src: my_custom_conn_id dest: datasource_name

再使用 –custom-rules 參數(shù)指定轉(zhuǎn)化自定義參數(shù)，就能應(yīng)用自定義規(guī)則的轉(zhuǎn)化：

# 指定自定義規(guī)則路徑為 /path/to/MyCustomOperator.yamlair2phin migrate --inplace --custom-rules /path/to/MyCustomOperator.yaml ~/airflow/dags

• 讓 Air2phin 運(yùn)行地更快

Air2phin 默認(rèn)是一個(gè)進(jìn)程運(yùn)行 DAG 文件的轉(zhuǎn)化的，當(dāng)你有許多 DAG 文件時(shí)，Air2phin 轉(zhuǎn)化非常耗時(shí)，我們提供了一個(gè)啟動(dòng)多進(jìn)程運(yùn)行 Air2phin 轉(zhuǎn)化的參數(shù) –multiprocess，可以將其指定為用戶機(jī)器的 CPU 數(shù)量來縮短轉(zhuǎn)化時(shí)間：

# 指定 air2phin 啟動(dòng) 12 個(gè)進(jìn)程同時(shí)進(jìn)行轉(zhuǎn)化air2phin migrate --inplace --custom-rules /path/to/MyCustomOperator.yaml --multiprocess 12 ~/airflow/dags

存在的問題

目前，作為一個(gè)轉(zhuǎn)化工具，Air2phin 的使用方式已經(jīng)算比較完善了，能夠滿足用戶遷移調(diào)度系統(tǒng)的基本需求，但還有一些地方有待完善。

內(nèi)置規(guī)則還不夠多

轉(zhuǎn)化規(guī)則還不夠多，目前只有五個(gè)，分別是：

• airflow.DAG

• airflow.operators.bash.BashOperator

• airflow.operators.dummy_operator.DummyOperator

• airflow.operators.python_operator.PythonOperator

• airflow.operators.spark_sql_operator.SparkSqlOperator

如果有更多的規(guī)則，Air2phin 將成為一個(gè)更加好用的轉(zhuǎn)化工具，這里歡迎各位隨時(shí)提交轉(zhuǎn)化規(guī)則的PR(https://github.com/WhaleOps/air2phin/pulls)。

部分Airflow的用法不能被遷移過來

部分概念僅僅在 Airflow 中有，在 DolphinScheduler 中還沒有，如任務(wù)的成功、失敗、重試、觸發(fā) callback，任務(wù)的 owner，variable，工作流并發(fā)數(shù)，tag 等，這部分 Airflow DAG 可以被遷移，但兼容的屬性將會(huì)丟失，無法遷移到 DolphinScheduler。

Air2phin 常見問題解答

Q：為什么選擇解析 Airflow DAG 文件而不是數(shù)據(jù)庫？

A：因?yàn)?Airflow DAG 文件中才有完成的工作流信息，Airflow 的數(shù)據(jù)庫中只有工作流基本信息，沒有任務(wù)定義的信息，也沒有任務(wù)的關(guān)系，我們選擇通過解析 Airflow 的 DAG 文件而不是數(shù)據(jù)庫來完成轉(zhuǎn)化。

Q：為什么要通過 dolphinscheduler python sdk 做中轉(zhuǎn)不自己提交到 DolphinScheduler？

A：因?yàn)?Airflow DAG 就是 Python 定義的，在 Airflow DAG 中有很多 Python 的特性，我們不想將這部分特性轉(zhuǎn)化成結(jié)構(gòu)化的數(shù)據(jù)（轉(zhuǎn)化可能存在信息丟失），恰好 DolphinScheduler 已經(jīng)有了 Python 的 sdk，所以直接通過 LibCST 轉(zhuǎn)化是成本更加低的做法。

Q：為什么使用 LibCST 而不是 python 內(nèi)置的 AST？

A：因?yàn)?LibCST 更加符合我們，Python 內(nèi)置的 AST 庫解析成 AST 的時(shí)候會(huì)丟失掉 comment 的信息，但是我們呢希望保留著部分信息。且 LibCST 提供更加多 visitor 保證我們更加方便的實(shí)現(xiàn)替換。

參考鏈接：air2phin(https://github.com/WhaleOps/air2phin)

?騰訊回應(yīng)進(jìn)軍類 ChatGPT；Meta 新語言模型能運(yùn)行在單張顯卡上；OpenAI 創(chuàng)始人提出新摩爾定律|極客頭條

?ChatGPT 帶火的「提示工程師」崗，不用寫代碼，也能獲得年薪數(shù)百萬？

?“ChatGPT 正在取代員工”，最新 ChatGPT 調(diào)查報(bào)告發(fā)布！