IOError:没有基于文件模式的文件

Question

我正在尝试运行Python中的示例。但是，这个堆栈跟踪的错误如下所示。注意:第一个管道确实创建了“./name”文件，但是第二个管道似乎无法从中读取。

No handlers could be found for logger "oauth2client.contrib.multistore_file"
Traceback (most recent call last):
  File "example.py", line 17, in <module>
    | 'save' >> beam.io.WriteToText(greetings_file))
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/io/textio.py", line 391, in __init__
    skip_header_lines=skip_header_lines)
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/io/textio.py", line 88, in __init__
    validate=validate)
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/io/filebasedsource.py", line 97, in __init__
    self._validate()
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/io/filebasedsource.py", line 173, in _validate
    'No files found based on the file pattern %s' % self._pattern)
IOError: No files found based on the file pattern ./names

示例代码如下：

import apache_beam as beam
def add_greeting(name, messages):
    for msg in messages:
        yield '%s %s' % (msg, name)

names_file = './names'
greetings_file = './greetings'

p = beam.Pipeline('DirectRunner')
(p | 'add names' >> beam.Create(['Ann', 'Joe'])
   | 'save' >> beam.io.WriteToText(names_file))
p.run()

(p
 | 'load names' >> beam.io.ReadFromText(names_file)
 | 'add greetings' >> beam.FlatMap(add_greetings, ['Hello', 'Hola'])
 | 'save' >> beam.io.WriteToText(greetings_file))
p.run()

环境:我在google云shell上运行这个

$ pip list --local --format=columns | grep dataflow
google-cloud-dataflow              0.6.0 

Answer 1

当管道运行时，Beam中的运行程序不会等待它完成，因此您应该在调用wait_until_finish()之后添加对p.run()的调用。

此外，Beam管道有延迟执行，因此当您为管道定义新步骤时，它们将被添加到每次运行管道时完全执行的图形中。简而言之，这意味着如果您想要运行不同步骤的管道，就需要创建一个新的Pipeline对象。

这应该是可行的：

p = beam.Pipeline('DirectRunner')
(p | 'add names' >> beam.Create(['Ann', 'Joe'])
   | 'save' >> beam.io.WriteToText('./names'))
p.run().wait_until_finish()

p = beam.Pipeline('DirectRunner')
(p
 | 'load names' >> beam.io.ReadFromText('./names*')
 | 'add greetings' >> beam.FlatMap(add_greeting, ['Hello', 'Hola'])
 | 'save' >> beam.io.WriteToText(greetings_file))
p.run().wait_until_finish()