Luigi任务不运行：执行TaskA时其依赖的TaskB、TaskC未触发运行

核心根因 #

• Luigi默认调度逻辑为优先检查目标任务的输出文件是否存在：如果指定运行的TaskA的输出文件./logs/task_a.txt已存在，Luigi会直接判定TaskA已完成，不会校验其依赖TaskB、TaskC的状态，也不会触发任何任务执行。
• 代码本身存在两处会导致执行失败的问题：
  • 未提前创建./logs目录，LocalTarget打开文件时会抛出路径不存在异常
  • outfile.write方法仅支持接收1个字符串参数，代码中用逗号分隔格式化字符串和变量的写法不符合语法要求，需要用%运算符或者f-string完成字符串格式化

修复方案 #

1. 清理历史输出 #

删除./logs目录下所有已生成的txt文件，或直接删除整个logs目录，避免Luigi误判任务已完成。

2. 修正代码 #

修改后的完整可运行代码如下：

import luigi
import os

count = 0

# 自动创建logs目录，避免路径不存在报错
if not os.path.exists('./logs'):
    os.makedirs('./logs')

class TaskC(luigi.Task):
    def requires(self):
        return None

    def run(self):
        print("Running task C ...")
        global count
        with self.output().open('w') as outfile:
            outfile.write("Finished task C, count = %d" % count)
            count += 1
        
    def output(self):
        return luigi.LocalTarget("./logs/task_c.txt")


class TaskB(luigi.Task):
    def requires(self):
        return None

    def run(self):
        print("Running task B ...")
        global count
        with self.output().open('w') as outfile:
            outfile.write("Finished task B, count = %d ..." % count)
            count += 1
        
    def output(self):
        return luigi.LocalTarget("./logs/task_b.txt")


class TaskA(luigi.Task):
    def requires(self):
        return [TaskB(), TaskC()]

    def run(self):
        print("Running task A ...")
        global count
        with self.output().open('w') as outfile:
            outfile.write("Finished task A, count = %d ..." % count)
            count += 1

    def output(self):
        return luigi.LocalTarget("./logs/task_a.txt")

if __name__ == '__main__':
    print("Start the first luigi app :)")
    luigi.run()

3. 重新执行任务 #

运行命令保持不变：

python first_luigi_app.py --local-scheduler TaskA

效果说明 #

修复后运行任务，Luigi会先检查到TaskA的输出不存在，递归校验依赖TaskB、TaskC的输出也不存在，依次执行两个依赖任务（同级依赖默认无固定执行顺序），两个依赖全部完成后再执行TaskA，三个任务的输出文件都会正常生成在logs目录下。

内容的提问来源于stack exchange，提问作者TngD