5

我最近开始使用Luigi，我想了解如何使用它来不断地将新数据附加到现有的目标文件中。

想象一下，我每分钟都在 ping 一个 api 来检索新数据。因为 aTask仅在Target不存在的情况下运行，所以一种天真的方法是通过当前的datetime. 这是一个简单的例子：

import luigi
import datetime

class data_download(luigi.Task):
    date = luigi.DateParameter(default = datetime.datetime.now()) 

    def requires(self):
        return []

    def output(self):
        return luigi.LocalTarget("data_test_%s.json" % self.date.strftime("%Y-%m-%d_%H:%M"))

    def run(self):
        data = download_data()
        with self.output().open('w') as out_file:
            out_file.write(data + '\n')

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

如果我安排这个任务每分钟运行一次，它将执行，因为当前时间的目标文件还不存在。但它每分钟创建 60 个文件。相反，我想做的是确保所有新数据最终都在同一个文件中。实现这一目标的可扩展方法是什么？欢迎任何想法，建议！

4

2 回答 2

1

你不能。正如文档LocalTarget所说：

参数： mode (str) – 模式 r 以只读模式打开 FileSystemTarget，而 w 将以写入模式打开 FileSystemTarget。子类可以实现其他选项。

即只允许r或w模式。其他选项，例如a需要扩展LocalTarget类；尽管它打破了 Luigi 任务执行所需的幂等性。

于 2019-01-15T15:50:41.997 回答
0 
def output(self):
        return luigi.LocalTarget("data_test_%s.json" % self.date.strftime("%Y-%m-%d_%H:%M"))

这不是“luigi 方式”，但它可以完成工作。最后，这些目标只是文件对象。

于 2017-08-08T19:02:56.893 回答