У меня есть две задачи Луиджи.
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.