大數(shù)據(jù)是復(fù)雜的，我已經(jīng)寫了很多關(guān)于廣闊的生態(tài)系統(tǒng)和廣泛的可用選項(xiàng)的文章。 通常被忽略但很關(guān)鍵的一個(gè)方面是管理大數(shù)據(jù)管道的不同步驟的執(zhí)行。 框架的決定或執(zhí)行過(guò)程的設(shè)計(jì)經(jīng)常會(huì)推遲到稍后的階段，從而導(dǎo)致許多問(wèn)題并延誤項(xiàng)目。

[[348345]]

您應(yīng)該盡早設(shè)計(jì)管道編排，以避免在部署階段出現(xiàn)問(wèn)題。 編排應(yīng)像其他可交付成果一樣對(duì)待; 所有利益相關(guān)者都應(yīng)該對(duì)其進(jìn)行計(jì)劃，實(shí)施，測(cè)試和審查。

編排框架通常被忽略，許多公司最終為其管道實(shí)施定制解決方案。 這不僅成本高昂，而且效率低下，因?yàn)樽远x業(yè)務(wù)流程解決方案往往會(huì)面臨現(xiàn)成框架已經(jīng)解決的相同問(wèn)題。 造成漫長(zhǎng)的反復(fù)試驗(yàn)。

在本文中，我將介紹一些最常見(jiàn)的開(kāi)源業(yè)務(wù)流程框架。

管道編排

數(shù)據(jù)管道編排是一個(gè)交叉過(guò)程，可管理管道任務(wù)之間的依賴關(guān)系，調(diào)度作業(yè)等。 如果使用流處理，則需要編排每個(gè)流應(yīng)用程序的依賴關(guān)系，而對(duì)于批處理，則需要安排和編排作業(yè)。

請(qǐng)記住，任務(wù)和應(yīng)用程序可能會(huì)失敗，因此您需要一種以統(tǒng)一的方式調(diào)度，重新調(diào)度，重放，監(jiān)視，重試和調(diào)試整個(gè)數(shù)據(jù)管道的方法。

業(yè)務(wù)流程框架提供的一些功能是：

• 作業(yè)調(diào)度
• 依賴管理
• 錯(cuò)誤管理和重試
• 工作參數(shù)化
• SLA跟蹤，警報(bào)和通知
• 具有儀表板的用戶界面，例如甘特圖和圖形
• 歷史和審計(jì)
• 元數(shù)據(jù)的數(shù)據(jù)存儲(chǔ)
• 日志匯總

讓我們回顧一下一些選項(xiàng)…

Apache Oozie

Apache Oozie是Hadoop的調(diào)度程序，作業(yè)創(chuàng)建為DAG，并且可以由基于cron的調(diào)度或數(shù)據(jù)可用性觸發(fā)。 Oozie是作為Java Web應(yīng)用程序運(yùn)行的可伸縮，可靠和可擴(kuò)展的系統(tǒng)。 它與Sqoop等提取工具和Spark等處理框架集成在一起。

Oozie工作流程定義以hPDL(XML)編寫。 工作流包含控制流節(jié)點(diǎn)和動(dòng)作節(jié)點(diǎn)。 控制流節(jié)點(diǎn)定義工作流的開(kāi)始和結(jié)束(開(kāi)始，結(jié)束和失敗節(jié)點(diǎn))，并提供一種機(jī)制來(lái)控制工作流的執(zhí)行路徑(決策，派生和聯(lián)接節(jié)點(diǎn))[1]。

動(dòng)作節(jié)點(diǎn)是一種機(jī)制，工作流通過(guò)該機(jī)制觸發(fā)任務(wù)的執(zhí)行。 Oozie支持不同類型的操作(map-reduce，Pig，SSH，HTTP，電子郵件…)，并且可以擴(kuò)展以支持其他類型的操作[1]。

同樣，可以對(duì)工作流程進(jìn)行參數(shù)設(shè)置，并且可以同時(shí)執(zhí)行幾個(gè)相同的工作流程作業(yè)。

它是Hadoop的第一個(gè)調(diào)度程序，非常流行，但是已經(jīng)有點(diǎn)過(guò)時(shí)了，如果您完全依賴Hadoop平臺(tái)，它仍然是一個(gè)不錯(cuò)的選擇。

Apache Airflow

Airflow是一個(gè)平臺(tái)，可用于計(jì)劃，運(yùn)行和監(jiān)視工作流程。 由于其易用性和創(chuàng)新的工作流作為代碼方法，它已成為大數(shù)據(jù)管道的最著名協(xié)調(diào)者，其中DAG在Python代碼中定義，可以像其他任何可交付的軟件一樣進(jìn)行測(cè)試。

它使用DAG創(chuàng)建復(fù)雜的工作流程。 圖中的每個(gè)節(jié)點(diǎn)都是一個(gè)任務(wù)，邊定義了任務(wù)之間的依賴關(guān)系。 任務(wù)分為兩類：

• 操作員：執(zhí)行一些操作。
• 傳感器：檢查過(guò)程或數(shù)據(jù)結(jié)構(gòu)的狀態(tài)。

Airflow Scheduler在遵循您描述的指定依賴項(xiàng)的同時(shí)，在一組工作線程上執(zhí)行您的任務(wù)。 它具有模塊化架構(gòu)，并使用消息隊(duì)列來(lái)協(xié)調(diào)任意數(shù)量的工作程序，并且可以擴(kuò)展到無(wú)窮大[2]。

它為您生成DAG，從而最大程度地提高了并行度。 DAG是用Python編寫的，因此您可以在本地運(yùn)行它們，對(duì)其進(jìn)行單元測(cè)試并將其與開(kāi)發(fā)工作流程集成。 當(dāng)工作流定義為代碼時(shí)，它們變得更加可維護(hù)，可版本控制，可測(cè)試和協(xié)作[2]。

豐富的用戶界面可以輕松地可視化生產(chǎn)中運(yùn)行的管道，監(jiān)視進(jìn)度并在需要時(shí)對(duì)問(wèn)題進(jìn)行故障排除[2]。 它快速，易于使用且非常有用。 它具有多種視圖和多種方法來(lái)解決問(wèn)題。 它保留了運(yùn)行的歷史記錄，以供以后參考。

> Airflow UI[2]: https://airflow.apache.org/docs/stable/

安裝非常簡(jiǎn)單。 您只需要Python。 它具有兩個(gè)獨(dú)立運(yùn)行的進(jìn)程，即UI和Scheduler。

原則[2]：

• 動(dòng)態(tài)的：氣流管道是通過(guò)代碼(Python)配置的，從而可以動(dòng)態(tài)生成管道。 這允許編寫可動(dòng)態(tài)實(shí)例化管道的代碼。
• 可擴(kuò)展：輕松定義您自己的運(yùn)算符，執(zhí)行程序并擴(kuò)展庫(kù)，使其適合于您的環(huán)境的抽象級(jí)別。
• 優(yōu)雅：氣流管道簡(jiǎn)潔明了。 使用強(qiáng)大的Jinja模板引擎將參數(shù)化腳本內(nèi)置到Airflow中。
• 可擴(kuò)展

盡管氣流是作為代碼編寫的，但是氣流并不是數(shù)據(jù)流解決方案[2]。 此外，工作流預(yù)計(jì)大部分是靜態(tài)的或緩慢變化的，對(duì)于非常小的動(dòng)態(tài)作業(yè)，還有其他選項(xiàng)，我們將在后面討論。

盡管XCOM功能用于在經(jīng)常需要的任務(wù)之間傳遞小的元數(shù)據(jù)，例如當(dāng)您需要某種相關(guān)性ID時(shí)，它卻是簡(jiǎn)單且無(wú)狀態(tài)的。 它還支持變量和參數(shù)化作業(yè)。 最后，它具有支持SLA和警報(bào)。 它可以與用于監(jiān)視的通話工具集成在一起。

Luigi是具有類似功能的Airflow的替代產(chǎn)品，但Airflow具有更多功能，并且比Luigi具有更好的擴(kuò)展性。

Dagster

Dagster是機(jī)器學(xué)習(xí)，分析和ETL的新編排者[3]。 主要區(qū)別在于，您可以像Apache NiFi一樣跟蹤數(shù)據(jù)的輸入和輸出，從而創(chuàng)建數(shù)據(jù)流解決方案。 這意味著它可以跟蹤執(zhí)行狀態(tài)，并可以將值具體化為執(zhí)行步驟的一部分。 您可以使用數(shù)據(jù)管道和資產(chǎn)的統(tǒng)一視圖在本地測(cè)試并在任何地方運(yùn)行。 它支持任何云環(huán)境。

Dagster對(duì)業(yè)務(wù)流程圖中各步驟之間的數(shù)據(jù)依賴關(guān)系進(jìn)行建模，并處理它們之間的數(shù)據(jù)傳遞。 輸入和輸出上的可選類型有助于盡早發(fā)現(xiàn)錯(cuò)誤[3]。 管道由共享的，可重用的，可配置的數(shù)據(jù)處理和基礎(chǔ)架構(gòu)組件構(gòu)建而成。 Dagster的網(wǎng)絡(luò)用戶界面使任何人都可以檢查這些對(duì)象并發(fā)現(xiàn)如何使用它們[3]。

> Dagster UI[4]: https://docs.dagster.io/

它還可以并行運(yùn)行多個(gè)作業(yè)，易于添加參數(shù)，易于測(cè)試，提供簡(jiǎn)單的版本控制，出色的日志記錄，故障排除功能等等。 與Airflow相比，它具有更多功能，但是它還有些不成熟，并且由于它需要跟蹤數(shù)據(jù)，因此可能難以擴(kuò)展，由于狀態(tài)性，這是NiFi面臨的一個(gè)問(wèn)題。 而且它很大程度上基于Python生態(tài)系統(tǒng)。

Prefect

Prefect與Dagster相似，提供本地測(cè)試，版本控制，參數(shù)管理等等。 它也是基于Python的。

Prefect之所以與眾不同，是為了克服Airflow執(zhí)行引擎的局限性，例如改進(jìn)的調(diào)度程序，參數(shù)化的工作流，動(dòng)態(tài)工作流，版本控制和改進(jìn)的測(cè)試。 對(duì)于許多面向DevOps的組織來(lái)說(shuō)，必須具有版本控制功能，但Airflow仍不支持版本控制，Prefect確實(shí)支持該功能。

它具有一個(gè)核心的開(kāi)源工作流管理系統(tǒng)以及一個(gè)完全不需要設(shè)置的云產(chǎn)品。 Prefect Cloud由GraphQL，Dask和Kubernetes支持，因此可以隨時(shí)使用[4]。 UI僅在云產(chǎn)品中可用。

Apache NiFi

Apache NiFi不是業(yè)務(wù)流程框架，而是更廣泛的數(shù)據(jù)流解決方案。 NiFi還可以安排作業(yè)，監(jiān)視，路由數(shù)據(jù)，警報(bào)等等。 它專注于數(shù)據(jù)流，但您也可以處理批處理。

它不需要任何類型的編程，并提供拖放UI。 它非常易于使用，您可以將其用于中等難度的作業(yè)，而不會(huì)出現(xiàn)任何問(wèn)題，但是對(duì)于較大的作業(yè)，它往往存在可伸縮性問(wèn)題。

它在Hadoop外部運(yùn)行，但可以觸發(fā)Spark作業(yè)并連接到HDFS / S3。

> NiFi UI[5]: https://nifi.apache.org/

用例

讓我們看一些例子…

• 我有一個(gè)舊的Hadoop集群，其Spark批處理作業(yè)的運(yùn)行速度很慢，您的團(tuán)隊(duì)符合Scala開(kāi)發(fā)人員的要求，而您的DAG并不太復(fù)雜。 在這種情況下，Ozzie是一個(gè)不錯(cuò)的選擇，因?yàn)樗峁┝擞?jì)劃Spark作業(yè)的簡(jiǎn)單方法。
• 我有許多具有復(fù)雜依賴關(guān)系的運(yùn)行緩慢的Spark作業(yè)，您需要能夠測(cè)試依賴關(guān)系并最大化并行性，您需要一個(gè)易于部署且提供大量故障排除功能的解決方案。 在這種情況下，Airflow是您最好的選擇。
• 我需要從許多來(lái)源實(shí)時(shí)獲取數(shù)據(jù)，您需要跟蹤數(shù)據(jù)沿襲，路由數(shù)據(jù)，豐富數(shù)據(jù)并能夠調(diào)試任何問(wèn)題。 這是您的BA所需要的實(shí)時(shí)數(shù)據(jù)流傳輸管道，他們沒(méi)有太多的編程知識(shí)。 在這種情況下，Apache NiFi是您最好的選擇，因?yàn)樗恍枰狿ython技能即可提供所需的所有功能。 如果您的團(tuán)隊(duì)具備Python技能，請(qǐng)考慮使用Dagster。
• 我想在云中創(chuàng)建實(shí)時(shí)和批處理管道，而不必?fù)?dān)心維護(hù)服務(wù)器或配置系統(tǒng)。 我需要一個(gè)快速，強(qiáng)大的解決方案來(lái)增強(qiáng)基于Python的分析團(tuán)隊(duì)的能力。 在這種情況下，請(qǐng)使用Prefect Cloud。
• 我有短暫的，瞬息萬(wàn)變的工作，要處理要跟蹤的復(fù)雜數(shù)據(jù)，我需要一種方法來(lái)解決問(wèn)題并快速進(jìn)行生產(chǎn)變更。 在這種情況下，請(qǐng)考慮Dagster。
• 我處理數(shù)百TB的數(shù)據(jù)，我有一個(gè)復(fù)雜的依賴項(xiàng)，我想自動(dòng)化我的工作流程測(cè)試。 對(duì)于這種情況，請(qǐng)使用Airflow，因?yàn)樗梢詳U(kuò)展，與許多系統(tǒng)交互并可以進(jìn)行單元測(cè)試。 Dagster或Prefect可能在此規(guī)模的數(shù)據(jù)上存在規(guī)模問(wèn)題。
• 我不確定我需要什么。 在這種情況下，請(qǐng)從Airflow開(kāi)始，因?yàn)樗亲钍軞g迎的選擇。

結(jié)論

我們似乎是一些最常見(jiàn)的業(yè)務(wù)流程框架。 如您所見(jiàn)，它們中的大多數(shù)將DAG用作代碼，因此您可以在將新的工作流程投入生產(chǎn)之前在本地進(jìn)行測(cè)試，調(diào)試管道并對(duì)其進(jìn)行正確的測(cè)試。 考慮本文討論的所有功能，并選擇最適合該工作的工具。

簡(jiǎn)而言之，如果您的需求只是編排不需要共享數(shù)據(jù)的獨(dú)立任務(wù)，并且/或者您的工作很慢，并且/或者您不使用Python，請(qǐng)使用Airflow或Ozzie。 對(duì)于需要數(shù)據(jù)沿襲和跟蹤的數(shù)據(jù)流應(yīng)用程序，請(qǐng)對(duì)非開(kāi)發(fā)人員使用NiFi; 或Dagster或Prefect(適用于Python開(kāi)發(fā)人員)。

在可能的情況下，請(qǐng)嘗試使工作保持簡(jiǎn)單并在Orchestrator外部管理數(shù)據(jù)依賴關(guān)系，這在Spark中很常見(jiàn)，在Spark中您將數(shù)據(jù)保存到深度存儲(chǔ)中而不傳遞。 在這種情況下，Airflow是一個(gè)不錯(cuò)的選擇，因?yàn)樗恍枰檾?shù)據(jù)流，并且您仍然可以使用XCOM傳遞小的元數(shù)據(jù)，例如數(shù)據(jù)的位置。 對(duì)于更小，運(yùn)行速度更快，基于python的作業(yè)或更多動(dòng)態(tài)數(shù)據(jù)集，您可能希望在Orchestrator中跟蹤數(shù)據(jù)依賴性并使用Dagster之類的工具。