在上一篇Luigi的线性调度文章中(Luigi任务调度框架学习1:线性调用流程),我们知道Task运行的时候:
每个任务是否完成有两次判定,即:进行判定(未完成) =》运行def run(self)函数 =》进行判定(完成) =》运行后续的Task;如果第一次判定就完成,则不会执行当前Task的def run(self)函数
但是Luigi支持的判定条件只有文件与SQL,在有些情况下(尤其是定时任务),我们希望它梳理并运行整个拓扑,而不是根据判定去决定是否运行,因此本文来解决这个问题,让我们能够直接运行Task
文章目录
- 解决方法
- 问题解析
- 1. Luigi多Task多次运行效果
- 2. 无限调用Task示例程序
解决方法
博主写了一个工具类:
import luigi
import os
class NeverStopLuigi(luigi.Task):
"""无限执行Task的工具类"""
never_stop = luigi.BoolParameter(True)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._out_put_file = self.__class__.__name__ # 用当前类名,作为无限重启确认的文件名
if self.never_stop:
outputs = luigi.task.flatten(self.output())
for out in outputs:
if isinstance(out, luigi.LocalTarget) and out.exists():
os.remove(self.output().path)
def run(self):
with open(self._out_put_file, "w") as f:
f.write(".")
def output(self):
return luigi.LocalTarget(self._out_put_file)
之后程序只需要继承NeverStopLuigi
类后覆写两个函数(def requires(self)
与def run(self)
)即可运行:
class MyTask(NeverStopLuigi):
def requires(self): # 1. 用于定义前置程序
print(f"运行MyTask的 requires程序")
return []
def run(self): # 2. 这个Task运行的主程序
super(MyTask, self).run() # 调用NeverStopLuigi的run函数
print(f"运行MyTask的 ======= 主程序 ======")
在这个工具类中,主要内容有:
- 依靠初始化Task时删除掉检测的文件,来实现首次检测output不通过
- 调用run函数先写文件,实现二次检测output通过
至于用于检测的文件名,用最不容易重名的类名self.__class__.__name__
来作为中间的文件名,这里是为了文件能够循环执行,而不用uuid这种随意生成的文件,最后无法删除(其实是可以在最后一个Task通过获取前面的所有output删除的,但没有直接继承的写法简洁明了)
问题解析
1. Luigi多Task多次运行效果
import luigi
class PreClass(luigi.Task):
out_file = "pre_class.text"
def requires(self): # 所需的前置函数,也可以不写,默认为空
print(f"运行PreClass的 requires程序")
return []
def output(self): # 完成的依据
print(f"运行PreClass的 output程序")
return luigi.LocalTarget(self.out_file) # 判断这一部分是否完成会去查看是否有out_file文件(pre_class.text)
def run(self): # 这一部分执行的主函数
print(f"运行PreClass的 ======= 主程序 ======")
with open(self.out_file, "a") as file:
file.write("啊哈哈哈")
class Run1Class(luigi.Task):
out_file = "run_1_class.text"
# 这里不覆写 output是默认空
def requires(self): # 所需的前置函数
print(f"运行Run1Class的 requires程序")
return [PreClass()]
def output(self): # 完成的依据
print(f"运行Run1Class的 output程序")
return luigi.LocalTarget(self.out_file) # 判断这一部分是否完成会去查看是否有out_file文件(pre_class.text)
def run(self): # 这一部分执行的主函数
print(f"运行Run1Class ======= 主程序 ======")
with open(self.out_file, "a") as file:
file.write("啊哈哈哈")
class Run2Class(luigi.Task):
out_file = "run_2_class.text"
# 这里不覆写 output是默认空
def requires(self): # 所需的前置函数
print(f"运行Run2Class的 requires程序")
return [Run1Class()]
def run(self): # 这一部分执行的主函数
print(f"运行Run2Class ======= 主程序 ======")
with open(self.out_file, "a") as file:
file.write("啊哈哈哈")
if __name__ == '__main__':
luigi.build([Run2Class()], local_scheduler=True) # 只写入最后输入的数据即可
最后的输出效果如下:
===== Luigi Execution Summary =====
Scheduled 3 tasks of which:
* 3 ran successfully:
- 1 PreClass()
- 1 Run1Class()
- 1 Run2Class()
This progress looks :) because there were no failed tasks or missing dependencies
===== Luigi Execution Summary =====
运行Run2Class的 requires程序
运行Run1Class的 output程序
运行Run1Class的 requires程序
运行PreClass的 output程序
运行PreClass的 requires程序
运行PreClass的 requires程序
运行PreClass的 ======= 主程序 ======
运行Run1Class的 requires程序
运行PreClass的 output程序
运行Run1Class ======= 主程序 ======
运行Run2Class的 requires程序
运行Run1Class的 output程序
运行Run2Class ======= 主程序 ======
调度顺序如上所示
2. 无限调用Task示例程序
import luigi
import os
class NeverStopLuigi(luigi.Task):
"""工具类"""
never_stop = luigi.BoolParameter(True)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._out_put_file = self.__class__.__name__ # 用当前类名,作为无限重启确认的文件名
if self.never_stop:
outputs = luigi.task.flatten(self.output())
for out in outputs:
if isinstance(out, luigi.LocalTarget) and out.exists():
os.remove(self.output().path)
def run(self):
with open(self._out_put_file, "w") as f:
f.write(".")
def output(self):
return luigi.LocalTarget(self._out_put_file)
class PreClass(NeverStopLuigi):
def requires(self): # 所需的前置函数,也可以不写,默认为空
print(f"运行PreClass的 requires程序")
return []
def run(self): # 这一部分执行的主函数
super(PreClass, self).run()
print(f"运行PreClass的 ======= 主程序 ======")
class Run1Class(NeverStopLuigi):
def requires(self): # 所需的前置函数
print(f"运行Run1Class的 requires程序")
return [PreClass()]
def run(self): # 这一部分执行的主函数
super(Run1Class, self).run()
print(f"运行Run1Class ======= 主程序 ======")
class Run2Class(NeverStopLuigi):
def requires(self): # 所需的前置函数
print(f"运行Run2Class的 requires程序")
return [Run1Class()]
def run(self): # 这一部分执行的主函数
super(Run2Class, self).run()
print(f"运行Run2Class ======= 主程序 ======")
if __name__ == '__main__':
luigi.build([Run2Class()], local_scheduler=True) # 只写入最后输入的数据即可
如上程序,不论运行多少次都会得到:
运行Run2Class的 requires程序
运行Run1Class的 requires程序
运行PreClass的 requires程序
运行PreClass的 requires程序
运行PreClass的 ======= 主程序 ======
运行Run1Class的 requires程序
运行Run1Class ======= 主程序 ======
运行Run2Class的 requires程序
运行Run2Class ======= 主程序 ======