python - 在 python luigi 上安排非常大量的工作

Question

我已经编写了一个 Luigi 管道来提取 1.2 mio 文件，然后对它们进行一些 sed 工作 - 参见 https://gist.github.com/wkerzendorf/395c85a2955002412be302d708329f7f .

如果我通过 Luigi 在几千个文件上运行它，它会安排得很好。但是在整个数据集上运行它会提示 Failed connecting to remote scheduler。不确定我这样做是否正确。

Answer 1

当文件超过 1k 时，我建议不要为每个文件创建单独的任务。您可能会更幸运地创建一个在这些文件的目录上运行的批处理任务。然后，此任务可以使用多处理来利用对您的处理函数的并行调用。

from multiprocessing import Pool, cpu_count
import os

class TestTask(luigi.WrapperTask):
    inglob = luigi.Parameter(default='/1002/*.gz')
    outdir = luigi.Parameter(default='/1002-out/')
    tmpdir = luigi.Parameter(default='/1002-tmp/'

    def extract_file(filename):
        # extract file to self.tempdir not shown

    def output(self):
        return luigi.LocalTarget(self.outdir)

    def run(self):
        os.makedirs(self.tempdir)
        p = Pool(cpu_count())
        p.map(extract_file, glob(self.inglob))
        os.rename(self.tempdir, self.outdir)