1

我有以下使用 Luigi Orchestrator 的 python 代码。

class AggregateArtists(luigi.Task):
    date = luigi.DateParameter(default=date.today() - timedelta(days=1))

    def requires(self):
        return []

    def run(self):
        ...

我想在我的run()函数中使用日期参数。问题是我不知道它是什么类型。在文档中，似乎这个参数是 a datetime.date，所以我应该可以使用方法self.date.strftime()。但此方法不适用于DateParameters.

我的问题是：

  • 如何在我的运行函数中使用我的代码的可变日期？它是什么类型的？一个字符串，一个 datetime.date 还是别的什么？

  • 在某些时候，我需要将此日期转换为 YYYYMMDD 形式的字符串，我该怎么做？

4

1 回答 1

1

您的代码不完整，但我猜其余部分如下所示。您必须在某处出现错误，因为它可以工作：DateParameter返回一个 python 日期的值。有关详细信息，请参阅luigi 源代码。

我的tasks/foo.py：

from datetime import date, timedelta
import luigi


class AggregateArtists(luigi.Task):
    date = luigi.DateParameter(default=date.today() - timedelta(days=1))

    def output(self):
        return luigi.LocalTarget("/tmp/foobar.txt")

    def run(self):
        with self.output().open('w') as out_file:
            out_file.write(self.date.strftime("%Y%m%d") + "\n")


if __name__ == "__main__":
    luigi.run()

运行任务：

$ python tasks/foo.py AggregateArtists --local-scheduler

DEBUG: Checking if AggregateArtists(date=2015-12-03) is complete
INFO: Scheduled AggregateArtists(date=2015-12-03) (PENDING)
INFO: Done scheduling tasks
INFO: Running Worker with 1 processes
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 21831] Worker Worker(salt=179482616, workers=1, host=matagus-laptop, username=matagus, pid=21831) running   AggregateArtists(date=2015-12-03)
INFO: [pid 21831] Worker Worker(salt=179482616, workers=1, host=matagus-laptop, username=matagus, pid=21831) done      AggregateArtists(date=2015-12-03)
DEBUG: 1 running tasks, waiting for next task to finish
DEBUG: Asking scheduler for work...
INFO: Done
INFO: There are no more tasks to run at this time
INFO: Worker Worker(salt=179482616, workers=1, host=matagus-laptop, username=matagus, pid=21831) was stopped. Shutting down Keep-Alive thread

打印输出文件的内容：

$ cat /tmp/foobar.txt 
20151203
于 2015-12-04T15:03:33.537 回答