大資料Hadoop工具python教程9-Luigi工作流
摘要：
管理Hadoop作業的官方工作流程排程程式是Apache Oozie。與許多其他Hadoop產品一樣，Oozie是用Java編寫的，是基於伺服器的Web應用程式，它執行執行Hadoop MapReduce和Pig的工作流作業。 Oozie工作流是在XML文件中指定的控制依賴性指導非迴圈圖(DA...
管理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能夠建立依賴鏈。
圖片.png
- 目標
目標是任務的輸入和輸出。最常見的目標是磁碟上的檔案，HDFS中的檔案或資料庫中的記錄。 Luigi包裝了底層檔案系統操作，以確保與目標的互動是原子的。這允許從故障點重放工作流，而不必重放任何已經成功完成的任務。
- 引數
引數允許通過允許值從命令列，以程式設計方式或從其他任務傳遞任務來自定義任務。例如，任務輸出的名稱可以通過引數傳遞給任務的日期來確定。
參考資料
- python測試開發專案實戰-目錄
- python工具書籍下載-持續更新
- python 3.7極速入門教程 - 目錄
- 討論qq群630011153 144081101
- 原文地址
- 本文涉及的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 luigi
class 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) + 1
ofp = 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 taskWordCount__home_hduser__in__tmp_wordcount2__a94efba0f2has statusPENDING INFO: Informed scheduler that taskInputFile__home_hduser__in_0eced493f7has statusDONE 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) runningWordCount(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) doneWordCount(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 taskWordCount__home_hduser__in__tmp_wordcount2__a94efba0f2has statusDONE 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 jack2 be2 nimble1 quick1