如何解决在 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)]
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。