如何定义inluigi.contrib.external\u program.ExternalProgramTas中的输出/完成

Question

我有两个任务，每个都是inluigi.contrib.external_program.ExternalProgramTask的子类

每个任务都在不同的dir上调用bash脚本，我想在它们之间添加依赖关系：

class ATask(ExternalProgramTask):
    def program_args(self):
        return my_script_a.split()

class BTask(ExternalProgramTask):
    def requires(self):
        return ATask()

    def program_args(self):
        return my_script_b.split()

if __name__ == '__main__':
    luigi.build([BTask()], workers=2, local_scheduler=False)

错误如下：

File .../local/lib/python2.7/site-packages/luigi/worker.py", line 182, in run
    raise RuntimeError('Unfulfilled %s at run time: %s' % (deps, ', '.join(missing)))
RuntimeError: Unfulfilled dependency at run time: ATask__...

据我所知-每个任务本身都是正常完成的（我可以在可视化工具中看到ATask正常完成，BTask输出成功运行所期望的结果），但是依赖性不知怎么搞砸了

据我所见，我需要定义一个output方法，并确保目标存在，比如添加：

def output(self):
    return luigi.LocalTarget('foo')

到ATask，但是，因为我不能重写这些任务中的run（因为它们是ExternalProgramTask子类），所以我不能创建那个文件（目标）。如何让BTask明白Atask已经完成了

Answer 1

在ATask中，添加complete（）函数，用于检查ATask是否已完成ok（通过返回True）或尚未完成ok（通过返回False）以及几个检查