Я написал конвейер Luigi для извлечения 1,2 млн файлов, а затем выполнил над ними некоторую работу с sed — см. //gist.github.com/wkerzendorf/395c85a2955002412be302d708329f7f.
Если я запускаю это через Луиджи на нескольких тысячах файлов, все работает нормально. Но запуская это на всем наборе данных, он жалуется на Failed connecting to remote scheduler. Не уверен, что я делаю это правильно.
Я бы посоветовал не создавать отдельную задачу для каждого файла, когда их будет больше 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)
