luigi批处理模块用于is直批处理任务

Question

我有500个链接下载，并想批处理它们，例如10个项目。在

这个伪代码是什么样子的？在

class BatchJobTask(luigi.Task)
    items = luigi.Parameter()
    def run(self):
        listURLs = []
        with ('urls_chunk', 'r') as urls
            for line in urls:
                listURLs.append('http://ggg'+line+'.org')
            10_urls = listURLs[0:items] #10 items here
            for i in 10_urls:
                req = request.get(url)
                req.contents
    def output(self):
        return self.LocalTarger("downloaded_filelist.txt")

class BatchWorker(luigi.Task)
    def run(self)
        # Here I should run BatchJobTask from 0 to 10, next 11 - 21 new etc...

会怎么样？在

Answer 1

我做到了。在

class GetListtask(luigi.Task)
    def run(self):
        ...
    def output(self):
    return luigi.LocalTarget(self.outputfile)

class GetJustOneFile(luigi.Task):
    fid = luigi.IntParameter()
    def requires(self):
        pass

    def run(self):
        url = 'http://my-server.com/test' + str(self.fid) + '.txt'
        download_file = requests.get(url, stream=True)
        with self.output().open('w') as downloaded_file:
            downloaded_file.write(str(download_file.content))

    def output(self):
        return luigi.LocalTarget("test{}.txt".format(self.fid))


class GetAllFiles(luigi.WrapperTask):
    def requires(self):
        listoffiles = []  # 0..999
        for i in range(899):
            listoffiles.append(i)
        return [GetJustOneFile(fid=fileid) for fileid in listoffiles]

这个代码很糟糕吗？在