Сбой конвейера при добавлении преобразования ReadAllFromText - PullRequest
Новая
       8

Сбой конвейера при добавлении преобразования ReadAllFromText

1 голос
/ 05 марта 2019

Я пытаюсь запустить очень простую программу в Apache Beam, чтобы проверить, как она работает. 

import apache_beam as beam


class Split(beam.DoFn):
    def process(self, element):
        return element


with beam.Pipeline() as p:
    rows = (p | beam.io.ReadAllFromText(
        "input.csv") | beam.ParDo(Split()))

При выполнении этого я получаю следующие ошибки 

.... some more stack....
 File "/home/raheel/code/beam-practice/lib/python2.7/site-packages/apache_beam/transforms/util.py", line 565, in expand
    windowing_saved = pcoll.windowing
  File "/home/raheel/code/beam-practice/lib/python2.7/site-packages/apache_beam/pvalue.py", line 137, in windowing
    self.producer.inputs)
  File "/home/raheel/code/beam-practice/lib/python2.7/site-packages/apache_beam/transforms/ptransform.py", line 464, in get_windowing
    return inputs[0].windowing
  File "/home/raheel/code/beam-practice/lib/python2.7/site-packages/apache_beam/pvalue.py", line 137, in windowing
    self.producer.inputs)
  File "/home/raheel/code/beam-practice/lib/python2.7/site-packages/apache_beam/transforms/ptransform.py", line 464, in get_windowing
    return inputs[0].windowing
AttributeError: 'PBegin' object has no attribute 'windowing'

Есть идеи, что здесь не так?

Спасибо

1 Ответ

3 голосов
/ 05 марта 2019

ReadAllFromText ожидает чтения из PCollection файлов вместо передачи его в качестве аргумента.Итак, в вашем случае это должно быть: 

p | beam.Create(["input.csv"])
  | beam.io.ReadAllFromText()
...