Luigi任务从未被调用，TaskA重复执行且yield TaskC后卡住问题

你的判断是正确的，触发该问题的核心原因有两个：

• Luigi的requires方法会在调度全周期（DAG构建、依赖检查、状态更新等阶段）被反复调用，并不是只执行一次。你看到两个分支的日志都被打印，是因为第一次调用requires时TaskB还未执行完成，S3上不存在对应输出所以进入if分支；等TaskB执行完毕写入S3后，Luigi再次调用requires检查依赖，此时就进入了else分支，并非同时进入两个分支。
• Luigi的静态DAG依赖树是在调度启动初期构建的，第一次调用TaskA.requires()时只返回了TaskB，TaskC没有被纳入依赖树。后续requires返回的新依赖不会被动态加入已有的调度队列，因此TaskC永远不会被执行，就会抛出「Task is never invoked」的报错，同时TaskA因为等不到依赖完成陷入无限卡住的状态。

另外你当前的写法还有一个典型错误：requires方法的唯一作用是声明任务依赖，不适合写入业务逻辑、结果读取、流程判断类的代码，这类代码应该放在run方法中执行。

你需要拆分独立任务、方便单独测试TaskC的需求可以通过如下方案实现：

方案1：静态声明依赖（推荐，适合固定依赖场景） #

直接在TaskC中声明对TaskB的依赖，这样单独调度TaskC时Luigi会自动先执行TaskB，无需耦合在TaskA的逻辑中，完全满足独立测试的需求。
修改后代码示例：

import luigi
from luigi.contrib.s3 import URITarget # 按需引入对应的URITarget实现

class TaskB(luigi.Task):
    def run(self): 
        # 处理逻辑并将结果写入S3
        pass

    def output(self):
        return URITarget('b_path')


class TaskC(luigi.Task):
    # 直接声明对TaskB的依赖，单独测试TaskC时可自动触发TaskB执行
    def requires(self):
        return TaskB()

    def run(self): 
        # 处理逻辑并将结果写入S3
        pass

    def output(self):
        return URITarget('c_path')


class TaskA(luigi.Task):
    # 直接声明依赖TaskC，Luigi会自动处理TaskB->TaskC->TaskA的执行顺序
    def requires(self):
        return TaskC()

    def run(self):
        # 所有业务逻辑、结果处理放在run中，仅会执行一次
        results = get_results_from_task_C_written_on_S3()
        # 执行其他业务逻辑
        pass

    def output(self):
        # 必须声明TaskA的输出，否则Luigi无法判断任务是否执行完成
        return URITarget('a_path')

方案2：动态依赖（适合需要条件触发TaskC的场景） #

如果你确实存在部分场景不需要执行TaskC的需求，可以使用Luigi的动态依赖特性，在TaskA的run方法中根据条件yield任务，此时依赖会被动态加入调度队列。
修改后代码示例：

class TaskA(luigi.Task):
    # 仅声明必须的前置依赖TaskB
    def requires(self):
        return TaskB()

    def run(self):
        # 所有条件判断放在run中，仅执行一次
        if not get_results_from_task_B_written_on_S3():
            print('Did not find any results and will exit')
            return
        print('Found results and will proceed')
        # 动态yield TaskC，Luigi会调度执行TaskC完成后再继续往下执行
        yield TaskC()
        results = get_results_from_task_C_written_on_S3()
        # 执行其他业务逻辑
        pass

    def output(self):
        return URITarget('a_path')

• 所有自定义的Luigi任务都需要声明output，否则Luigi无法判断任务是否执行成功，会出现重复调度的问题。
• 不要在requires、output这类会被反复调用的方法中写入有副作用的代码，所有单次执行的业务逻辑必须放在run方法中。

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