在 Apache Beam (python) 中使用 ToList 输出作为 AsSingleton 或 AsList 的输入

Question

当我尝试使用beam.combiners.ToList 的输出作为beam.pvalue.AsSingleton 或beam.pvalue.AsList 的输入以试验侧输入时出现意外错误。我能够使用单个数字（例如：列表的平均值）作为辅助输入，但是对于列表和字典，我得到了例外。对于beam.pvalue.AsSingleton，我得到：

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-4-0c1df7400a03> in <module>
     15 chain_total = chain_1 | chain_2
     16 
---> 17 chain_1 | beam.Map(m, beam.pvalue.AsList(chain_2))
     18 
     19 chain_total | beam.Map(print)

~/.cache/pypoetry/virtualenvs/prototyping-with-tensorflow-py3.6/lib/python3.6/site-packages/apache_beam/pvalue.py in __init__(self, pcoll)
    297     self.pvalue = pcoll
    298     self._window_mapping_fn = sideinputs.default_window_mapping_fn(
--> 299         pcoll.windowing.windowfn)
    300 
    301   def _view_options(self):

AttributeError: '_ChainedPTransform' object has no attribute 'windowing'

对于beam.pvalue.AsList，我得到：

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-7-0c1df7400a03> in <module>
     15 chain_total = chain_1 | chain_2
     16 
---> 17 chain_1 | beam.Map(m, beam.pvalue.AsList(chain_2))
     18 
     19 chain_total | beam.Map(print)

~/.cache/pypoetry/virtualenvs/prototyping-with-tensorflow-py3.6/lib/python3.6/site-packages/apache_beam/pvalue.py in __init__(self, pcoll)
    297     self.pvalue = pcoll
    298     self._window_mapping_fn = sideinputs.default_window_mapping_fn(
--> 299         pcoll.windowing.windowfn)
    300 
    301   def _view_options(self):

AttributeError: '_ChainedPTransform' object has no attribute 'windowing'

这是我正在运行的代码

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-4-0c1df7400a03> in <module>
     15 chain_total = chain_1 | chain_2
     16 
---> 17 chain_1 | beam.Map(m, beam.pvalue.AsList(chain_2))
     18 
     19 chain_total | beam.Map(print)

~/.cache/pypoetry/virtualenvs/prototyping-with-tensorflow-py3.6/lib/python3.6/site-packages/apache_beam/pvalue.py in __init__(self, pcoll)
    297     self.pvalue = pcoll
    298     self._window_mapping_fn = sideinputs.default_window_mapping_fn(
--> 299         pcoll.windowing.windowfn)
    300 
    301   def _view_options(self):

AttributeError: '_ChainedPTransform' object has no attribute 'windowing'

用 beam.pvalue.AsList 替换 beam.pvalue.AsSingleton 得到另一个错误。我正在使用 Apache Beam python SDK 2.11.0 版。

Answer 1

PCollections 是 Beam 中的名词，PTransforms 是动词。当您开始管道时，p = beam.Pipeline()是您唯一的名词。（注意 EvenCreate是一个动词。）

通过在该名词上应用各种动词，您可以根据以下规则创建其他名词：

• new_noun = existing_noun | verb

混乱的主要根源似乎是因为你也可以将动词链接在一起：

• fancy_verb = verb1 | verb2

虽然这些示例中的语法看起来非常相似，但返回的值具有不同的类型。

这里的主要问题是只有名词可以被视为 SideInputs。

在您提供的示例中，chain_2是通过组合两个动词创建的动词，并且错误消息确认_ChainedPTransform（实际上是一种PTransform，顾名思义）无法传递给任何 AsSideInput 函数。