使用Luigi管道时组织文件？

Question

我正在使用Luigi来完成我的工作流程。我的工作流程分为三个通用部分 - 导入，分析，导出。在每个部分中，有多个Luigi任务。

我可以将所有内容都放在一个文件中。但是，如果我想保持一切分开，就像有data_import.py，analysis.py和export.py。

例如，如果data_import.py看起来像：

import luigi

class import_task_A(luigi.Task):
    def requires(self):
        return []
    def output(self):
        return luigi.LocalTarget('myfile.txt')
    def run(self):
        my import stuff

if __name__ == '__main__':
    luigi.run()

但是，如果export.py中的任务依赖于import.py中的任务，该怎么办？我会这样做：

from data_import import import_task_A
import luigi

class export_task_A(luigi.Task):
    def requires(self):
        return import_task_A()
    def output(self):
        return luigi.LocalTarget('myfile.txt')
    def run(self):
        my import stuff

if __name__ == '__main__':
    luigi.run()

如果我将大型项目分解为多个.py文件，那么告诉Luigi哪个需要的任务在哪个文件中的最佳方法是什么？似乎这种方法会变得很麻烦。

Answer 1

为什么会变得麻烦？如果您的export_task_A依赖于许多任务，则您的def要求将更改为：

def requires(self):
    return [import_task_A(), import_task_B()]

顺便说一下，在这种情况下你可能想删除

if __name__ == '__main__':
    luigi.run()

来自您的data_import.py。而不是在data_export.py中使用相同的

if __name__ == '__main__':
    luigi.build([export_task_A()])

Answer 2

不确定是否有办法解决这个问题。您需要在一个文件中包含许多文件或许多类。您希望如何组织项目是一个优先事项。

你可以做的一件事是限制你导入的位置数是一个python文件导入你需要的所有Luigi类

# my_tasks.py
from data_import import import_task_A
from export import export_task_A

然后在其他文件中，您可以从my_tasks导入所需的任何内容。还可以考虑使用getattr或importlib来更灵活地导入和访问类。