как определить вывод / заполнить inluigi.contrib.external_program.ExternalProgramTask - PullRequest
Новая
       15

как определить вывод / заполнить inluigi.contrib.external_program.ExternalProgramTask

0 голосов
/ 13 мая 2019

У меня есть 2 задачи, каждая из которых является подклассом inluigi.contrib.external_program.ExternalProgramTask.

Каждая задача вызывает скрипт bash для разных директорий, и я хотел добавить зависимость между ними: 

class ATask(ExternalProgramTask):
    def program_args(self):
        return my_script_a.split()

class BTask(ExternalProgramTask):
    def requires(self):
        return ATask()

    def program_args(self):
        return my_script_b.split()

if __name__ == '__main__':
    luigi.build([BTask()], workers=2, local_scheduler=False)

Ошибка заключается в следующем: 

File .../local/lib/python2.7/site-packages/luigi/worker.py", line 182, in run
    raise RuntimeError('Unfulfilled %s at run time: %s' % (deps, ', '.join(missing)))
RuntimeError: Unfulfilled dependency at run time: ATask__...

Из того, что я понимаю - каждая задача завершается нормально сама по себе (я вижу ATask завершена нормально в визуализаторе, а BTask выводит то, что ожидается от успешного запуска), но зависимость как-то испортилась.

Из того, что я увидел, мне нужно определить метод output и убедиться, что цель будет существовать, что-то вроде добавления: 

def output(self):
    return luigi.LocalTarget('foo')

до ATask, НО , поскольку я не могу переопределить run в этих задачах (поскольку они являются ExternalProgramTask подклассами), я не могу создать этот файл (цель). Как я могу заставить BTask понять, что Atask закончил нормально?

1 Ответ

0 голосов
/ 13 мая 2019

в ATask, добавьте функцию complete (), которая проверяет, закончил ли ATask нормально (возвращая True) или не завершил нормально (возвращая False), и несколько проверок

...