[雪峰磁针石博客]大数据Hadoop工具python教程9-Luigi工作流
管理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 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 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
