Пользовательское задание Python для Beam / Data Flow - облачное хранилище в PubSub - PullRequest
Новая
       77

Пользовательское задание Python для Beam / Data Flow - облачное хранилище в PubSub

2 голосов
/ 03 июля 2019

Мне нужно выполнить очень простое преобразование некоторых данных (извлечь строку из 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
...