在 Luigi 中创建动态顺序任务

问题描述

import luigi

class FiletoStaging(ImportToTable):
    filename = luigi.Parameter(default = '')
    #import file from some folder to a staging database
    def requires(self):
         return luigi.LocalTarget(self.filename)
    
    #truncate table
    #load the file into staging


class StgToOfficial(RunQuery):
    filename = luigi.Parameter
    # run a process in the database to load data from staging to the final table
    def requires(self):
        return FiletoStaging(self.filename)

    # run query


class LoadFileGroups(luigi.WrapperTask):

    def requires(self):
        list_of_files = get_list_of_files_currently_in_folder() # The folder can have an arbitrary number of files inside
        for file in list_of_files:
            yield(StgToOfficial(filename = file))

你好，社区，

我是 luigi 的新手，正在尝试使用该框架构建 ETL 过程。

想象一下，我有一个类似于前面的伪代码片段的过程。该过程必须检查文件夹并获取其中的文件列表。然后，一一导入staging数据库，并运行一个进程将staging中的数据加载到最终表中。

问题在于，在之前的解决方案中，所有加载到暂存表中的文件（随后是每个文件的加载过程）都是并行运行的，这是不可能发生的。如何强制 luigi 按顺序执行任务？只有当一个文件在最终表中完成加载时，才导入下一个，依此类推。 （查看下面的草稿以获得简化的草稿）

Draft of the structure I'm trying to achieve

我知道我应该使用 requires 方法来确保顺序，但是对于要加载的未知数量的文件，我该如何动态地做到这一点？

非常感谢您的帮助。

解决方法

根据 Peter Weissbrod 在以下讨论中的回答，通过在 requires() 方法中创建递归模式来解决： https://groups.google.com/g/luigi-user/c/glvU_HxYmr0/m/JvV3xgsiAwAJ

这是彼得提出的解决方案：

这里是一个片段，它不是完全根据您的需求量身定制的，但同时您可能会发现它很有用。考虑这样的“sql”目录：

  ~ :>ll somedir/sql
-rw-rw-r-- 1 pweissbrod authenticatedusers   513 Jan 27 09:15 stage01_test.sqltemplate
-rw-rw-r-- 1 pweissbrod authenticatedusers  1787 Jan 28 13:57 stage02_test.sqltemplate
-rw-rw-r-- 1 pweissbrod authenticatedusers  1188 Jan 28 13:57 stage03_test.sqltemplate
-rw-rw-r-- 1 pweissbrod authenticatedusers 13364 Jan 29 07:16 stage04_test.sqltemplate
-rw-rw-r-- 1 pweissbrod authenticatedusers  1344 Jan 28 13:57 stage05_test.sqltemplate
-rw-rw-r-- 1 pweissbrod authenticatedusers  1983 Jan 28 17:03 stage06_test.sqltemplate
-rw-rw-r-- 1 pweissbrod authenticatedusers  1224 Jan 28 16:05 stage07_test.sqltemplate

# Consider a luigi task with a dynamic requires method like this:
class BuildTableTask(luigi.Task):
    table = luigi.Parameter(description='the name of the table this task (re)builds')
    def requires(self):
        tables = [f.split('_')[0] for f in os.listdir('sql') if re.match(f'stage[0-9]+[a-z]*_{config().environment}',f)]
        prereq = next(iter([t for t in sorted(tables,reverse=True) if t < self.table]),None)
        yield BuildTableTask(table=prereq) or []
   
    def run(self):
        with open(f'sql/{self.table}_{config().environment}.sqltemplate'.format(**config().to_dict())) as sqltemplate:
            sql = sqltemplate.read().format(**config().to_dict())
        db.run(f'create table {config().srcdbname}.{self.table} stored as orc as {sql}')

the task tree is built by observing the files in that directory:
 └─--[BuildTableTask-{'table': 'stage07'} (COMPLETE)]
    └─--[BuildTableTask-{'table': 'stage06'} (COMPLETE)]
       └─--[BuildTableTask-{'table': 'stage05'} (COMPLETE)]
          └─--[BuildTableTask-{'table': 'stage04'} (COMPLETE)]
             └─--[BuildTableTask-{'table': 'stage03'} (COMPLETE)]
                └─--[BuildTableTask-{'table': 'stage02'} (COMPLETE)]
                   └─--[BuildTableTask-{'table': 'stage01'} (COMPLETE)]