[雪峰磁针石博客]大数据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 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