使用 Luigi 管道时组织文件？

Posted 2023-02-19

【中文标题】使用 Luigi 管道时组织文件？

【英文标题】：Organizing files when using Luigi pipeline?

【发布时间】：2017-03-02 20:58:38

【问题描述】：

我在我的工作流程中使用 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 哪些必需任务在哪个文件中的最佳方法是什么？好像这个方法会比较麻烦。

【问题讨论】：

你的做法看起来不错。

为什么会变得很麻烦？

【参考方案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()])

【参考方案2】：

不确定有没有办法解决这个问题。您要么需要有许多文件，要么需要在一个文件中有许多类。您希望如何组织项目是一个偏好问题。

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

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

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