管理Hadoop作業的官方工作流程調度程序是Apache Oozie。與許多其他Hadoop產品一樣，Oozie是用Java編寫的，是基于服務器的Web應用程序，它運行執行Hadoop MapReduce和Pig的工作流作業。 Oozie工作流是在XML文檔中指定的控制依賴性指導非循環圖(DAG)中排列的動作集合。雖然Oozie在Hadoop社區中有很多支持，但通過XML屬性配置工作流和作業的學習曲線非常陡峭。

Luigi是Spotify創建的Python替代方案，可以構建和配置復雜的批處理作業管道。它處理依賴項解析，工作流管理，可視化等等。它還擁有龐大的社區，并支持許多Hadoop技術。在github上超過1萬星。

本章介紹Luigi的安裝和工作流程的詳細說明。

安裝

pip install luigi

工作流

在Luigi中，工作流由一系列操作組成，稱為任務。 Luigi任務是非特定的，也就是說，它們可以是任何可以用Python編寫的東西。任務的輸入和輸出數據的位置稱為目標(target)。目標通常對應于磁盤上，HDFS上或數據庫中的文件位置。除了任務和目標之外，Luigi還利用參數來自定義任務的執行方式。

• 任務

任務是構成Luigi工作流的操作序列。每個任務都聲明其依賴于其他任務創建的目標。這樣Luigi能夠創建依賴鏈。

• 目標

目標是任務的輸入和輸出。最常見的目標是磁盤上的文件，HDFS中的文件或數據庫中的記錄。 Luigi包裝了底層文件系統操作，以確保與目標的交互是原子的。這允許從故障點重放工作流，而不必重放任何已經成功完成的任務。

• 參數

參數允許通過允許值從命令行，以編程方式或從其他任務傳遞任務來自定義任務。例如，任務輸出的名稱可以通過參數傳遞給任務的日期來確定。

參考資料

• python測試開發項目實戰-目錄
• python工具書籍下載-持續更新
• python 3.7極速入門教程 - 目錄
• 原文地址
• 本文涉及的python測試開發庫 謝謝點贊！
• [本文相關海量書籍下載](https://github.com/china-testing/python-api-tesing/blob/master/books.md

工作流本示例

#!/usr/bin/env python # 項目實戰討論QQ群630011153 144081101 # https://github.com/china-testing/python-api-tesing import luigiclass InputFile(luigi.Task):"""A task wrapping a Target """input_file = luigi.Parameter()def output(self):"""Return the target for this task"""return luigi.LocalTarget(self.input_file)class WordCount(luigi.Task):"""A task that counts the number of words in a file"""input_file = luigi.Parameter()output_file = luigi.Parameter(default='/tmp/wordcount')def requires(self):"""The task's dependencies:"""return InputFile(self.input_file)def output(self):"""The task's output"""return luigi.LocalTarget(self.output_file)def run(self):"""The task's logic"""count = {}ifp = self.input().open('r')for line in ifp:for word in line.strip().split():count[word] = count.get(word, 0) + 1ofp = self.output().open('w')for k, v in count.items():ofp.write('{}\t{}\n'.format(k, v))ofp.close()if __name__ == '__main__':luigi.run()

執行

$ python wordcount.py WordCount --local-scheduler --input-file /home/hduser_/input2.txt --output-file /tmp/wordcount2.txt DEBUG: Checking if WordCount(input_file=/home/hduser_/input2.txt, output_file=/tmp/wordcount2.txt) is complete DEBUG: Checking if InputFile(input_file=/home/hduser_/input2.txt) is complete INFO: Informed scheduler that task WordCount__home_hduser__in__tmp_wordcount2__a94efba0f2 has status PENDING INFO: Informed scheduler that task InputFile__home_hduser__in_0eced493f7 has status DONE INFO: Done scheduling tasks INFO: Running Worker with 1 processes DEBUG: Asking scheduler for work... DEBUG: Pending tasks: 1 INFO: [pid 21592] Worker Worker(salt=067173106, workers=1, host=andrew-PC, username=hduser_, pid=21592) running WordCount(input_file=/home/hduser_/input2.txt, output_file=/tmp/wordcount2.txt) INFO: [pid 21592] Worker Worker(salt=067173106, workers=1, host=andrew-PC, username=hduser_, pid=21592) done WordCount(input_file=/home/hduser_/input2.txt, output_file=/tmp/wordcount2.txt) DEBUG: 1 running tasks, waiting for next task to finish INFO: Informed scheduler that task WordCount__home_hduser__in__tmp_wordcount2__a94efba0f2 has status DONE DEBUG: Asking scheduler for work... DEBUG: Done DEBUG: There are no more tasks to run at this time INFO: Worker Worker(salt=067173106, workers=1, host=andrew-PC, username=hduser_, pid=21592) was stopped. Shutting down Keep-Alive thread INFO: ===== Luigi Execution Summary =====Scheduled 2 tasks of which: * 1 complete ones were encountered:- 1 InputFile(input_file=/home/hduser_/input2.txt) * 1 ran successfully:- 1 WordCount(input_file=/home/hduser_/input2.txt, output_file=/tmp/wordcount2.txt)This progress looks :) because there were no failed tasks or missing dependencies===== Luigi Execution Summary =====hduser_@andrew-PC:/home/andrew/code/HadoopWithPython/python/Luigi$ cat /tmp/wordcount2.txt jack 2 be 2 nimble 1 quick 1

總結

以上是生活随笔為你收集整理的[雪峰磁针石博客]大数据Hadoop工具python教程9-Luigi工作流...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。