Luigi : Rangehourly示例

Question

是否有可用于RangeHourly提供的示例(或类似的示例，如RangeDaily)。我一直试图用它来重复执行任务。但我最终总是会犯这样的错误：

调试:检查RangeHourly(of=FinalTask，of_params={}，reverse=False，task_limit=50，now=None，param_name=None，start=2017-06-28T15，stop=None，hours_back=0，hours_forward=0)是否完成调试:空范围。没有FinalTask实例期望

以下是任务的定义：

class FinalTask (luigi.Task):
    start = luigi.DateHourParameter()
    def requires(self):
            return CleanupTask()
    def run(self):
            cmd='echo "Workflow Completed"'
            args=shlex.split(cmd)
            exc=subprocess.Popen(args,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
            stdout,stderr=exc.communicate()
            self.output().open('w').close()
    def output(self):
            return luigi.LocalTarget('/var/flags/FinalTask_success_%s.csv' %start)

有什么是我遗漏的，导致了这个问题吗？

Answer 1

python tasks.py RangeHourlyBase --of FinalTask --start 2017-07-31T00 --stop 2017-07-31T23 --local-scheduler --workers 4

请注意：

• 必须将tasks.py替换为定义FinalTask的文件名。
• --local-scheduler只用于在本地运行任务。不要用它“生产”。
• 最后一行有一个错误:没有定义start，应该是： 返回luigi.LocalTarget('/var/flags/FinalTask_success_%s.csv‘% self.start)