如何在Luigi工作流中实现条件分支并按需跳过对应子任务

原方案失效原因 #

Luigi的调度逻辑是在任务启动前就会完整遍历整棵依赖树，预检查所有任务的输出是否存在，提前确定待执行任务队列。你在Branch任务的run方法中才创建待跳过任务的输出文件，此时调度队列已经生成，所以该逻辑不会生效。

正确实现思路 #

不需要在分支任务中预先创建其他任务的输出文件，仅需要在分支后的合并节点动态读取分支任务的判断结果，按需返回需要执行的子任务即可，Luigi会自动调度对应任务，未被返回的任务自然不会被执行。

完整可运行代码示例 #

import luigi
from datetime import datetime

# 这里补充你原有Sleep任务的空实现，可替换为自己的逻辑
class Sleep(luigi.Task):
    def output(self):
        return luigi.LocalTarget('sleep.txt')
    def run(self):
        with self.output().open('w') as f:
            f.write('done')

class Branch(luigi.Task):
    def requires(self):
        return Sleep()

    def output(self):
        return luigi.LocalTarget('condition.txt')

    def run(self):
        hour = datetime.now().hour
        with self.output().open('w') as out:
            out.write('morning' if hour < 12 else 'afternoon')

class Morning(luigi.Task):
    def requires(self):
        return Branch()

    def output(self):
        return luigi.LocalTarget('morning.txt')

    def run(self):
        print("Morning")
        with self.output().open('w') as out:
            out.write("Morning")

class Afternoon(luigi.Task):
    def requires(self):
        return Branch()

    def output(self):
        return luigi.LocalTarget('afternoon.txt')

    def run(self):
        print("Afternoon")
        with self.output().open('w') as out:
            out.write("Afternoon")

class Merge(luigi.WrapperTask):
    def requires(self):
        # 先保证分支判断任务执行完成
        yield Branch()
        # 读取分支结果动态返回需要执行的子任务
        with Branch().output().open('r') as f:
            condition = f.read().strip()
        if condition == 'morning':
            yield Morning()
        else:
            yield Afternoon()

    def run(self):
        print("Merged")

效果说明 #

执行Merge任务时，Luigi会先运行Branch任务得到判断结果，再根据结果仅调度Morning或Afternoon中的一个任务执行，另一个任务不会进入执行队列，完全符合你需要的条件分支效果，对应逻辑和你参考的Prefect实现效果一致：

内容的提问来源于stack exchange，提问作者Gaëtan