从Beam中的另一个管道读取泡菜？

Question

我在谷歌云数据流中运行批处理管道。我需要在一个管道中读取另一个管道以前写过的对象。最简单的wa对象是pickle/dill。

编写工作很好，编写了许多文件，每个文件都有一个pickled对象。当我手动下载文件时，我可以打开文件。编写代码:beam.io.writeToText('gs://{}‘,coder=coders.dillcoder())

要么...

  File "/usr/local/lib/python2.7/dist-packages/dill/dill.py", line 266, in load
    obj = pik.load()
  File "/usr/lib/python2.7/pickle.py", line 858, in load
    dispatch[key](self)
KeyError: '\x90'

...或者...

  File "/usr/local/lib/python2.7/dist-packages/dill/dill.py", line 423, in find_class
    return StockUnpickler.find_class(self, module, name)
  File "/usr/lib/python2.7/pickle.py", line 1124, in find_class
    __import__(module)
ImportError: No module named measur

（对象的类位于带有measure的路径中，但不确定为什么会错过最后一个字符）

谢谢！

Answer 1

ReadFromText被设计为读取文本文件中新的分行记录，因此不适合您的用例。实现FileBasedSource也不是一个好的解决方案，因为它是为读取包含多条记录的大文件而设计的（并且通常将这些文件拆分为碎片进行并行处理）。因此，在您的情况下，Python SDK当前的最佳解决方案是自己实现源代码。这可以简单到读取文件并生成记录的pcollection的pardo。如果pardo生成大量记录，请考虑在后面添加apache_beam.transforms.util.reshuffle步骤，这将允许运行者更好地并行以下步骤。对于Java SDK，我们有fileio，它已经提供了转换，使这一点变得更容易。