Мне нужно выполнить очень простое преобразование некоторых данных (извлечь строку из JSON), а затем записать его в PubSub - я пытаюсь использовать для этого пользовательское задание потока данных Python.
Iя написал задание, которое успешно записывает обратно в облачное хранилище, но мои попытки даже самой простой записи в PubSub (без преобразования) приводят к ошибке: JOB_MESSAGE_ERROR: Workflow failed. Causes: Expected custom source to have non-zero number of splits.
Кто-нибудь успешно записал в PubSub из GCSчерез Dataflow?
Может кто-нибудь пролить свет на то, что здесь происходит не так?
def run(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument('--input',
dest='input',
help='Input file to process.')
parser.add_argument('--output',
dest='output',
help='Output file to write results to.')
known_args, pipeline_args = parser.parse_known_args(argv)
pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = True
with beam.Pipeline(options=pipeline_options) as p:
lines = p | ReadFromText(known_args.input)
output = lines #Obviously not necessary but this is where my simple extract goes
output | beam.io.WriteToPubSub(known_args.output) # This doesn't