在 Luigi 中创建动态顺序任务答案

【问题标题】：Create dynamically sequential tasks in Luigi在 Luigi 中创建动态顺序任务
【发布时间】：2021-05-16 09:39:39
【问题描述】：

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 流程。

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

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

Draft of the structure I'm trying to achieve

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

非常感谢您的帮助。

【问题讨论】：

标签： python dynamic etl sequential luigi

【解决方案1】：

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

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

这里有一个 sn-p，它不能完全满足您的需求，但同时您可能会发现它很有用。考虑一个像这样的“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)]

【讨论】：