Как использовать вывод задания в luigi - PullRequest
0 голосов
/ 29 января 2020

У меня есть две задачи Луиджи.

TaskA запускает внешнюю программу, и результаты сохраняются как json с luigi.LocalTarget

    class TaskA(ExternalProgramTask):
        def output(self):
           return luigi.LocalTarget(self.outputfile)

В TaskB я хочу выполнить некоторое преобразование json и pickle.dump() ит. Но у меня есть проблемы с открытием файла.

@inherits(TaskA)
class TaskB(luigi.Task):
    def requires(self):
        args = {....}
        return TaskA(**args)

    def run(self):
        try: 
            entries = json.load(self.input().open())
        except json.decoder.JSONDecodeError as e:
            logging.error(f"Decoding error: {e}")
            return print(e)

Но это не работает, так как я получаю ошибку декодирования:

Decoding error: Expecting value: line 6 column 1

, что имеет смысл, если Я пытаюсь напечатать self.input().open() Я ожидал бы json. Но вместо этого я получаю: <_io.TextIOWrapper name='task1.output.json' mode='rb' encoding='UTF-8'>

Я также пытался использовать yield TaskA(), но это также не работало. Но согласно документации, это должно работать.

Использование Python 3.8.1 и последней версии luigi.

1 Ответ

0 голосов
/ 30 января 2020

Я нашел проблему.

Действительно, ошибка декодирования имеет смысл. Я проверял свой ввод. json, который был сгенерирован внешним инструментом, и это был json.

Однако в инструменте была ошибка, и в последнюю запись добавлялась запятая, что делало его недействительным json.

...