如何从 Apache Beam 中的 Pcollection 中获取一个元素答案

【问题标题】：How to get one element from a Pcollection in Apache Beam如何从 Apache Beam 中的 Pcollection 中获取一个元素
【发布时间】：2023-04-10 21:18:01
【问题描述】：

考虑 Pcollection 的列表：

[{'id':'1','name':'Tom','country':'USA'},{'id':'2','name':'Oprah', '国家':'美国'}....]

我想统计每个国家的出现次数。结果应该是这样的：

{'USA':2, 'Tunisia':3, 'France':1}

【问题讨论】：

标签： python google-cloud-dataflow apache-beam

【解决方案1】：

检查beam.combiners.ToDict，结果会产生一个dict；

例子：

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

p = beam.Pipeline(options=PipelineOptions()) 

(p  
| "create pcoll" >> beam.Create([{'id':'1','name':'Tom','country':'USA'},
                                                {'id':'2','name':'Oprah','country':'USA'},
                                                {'id':'2','name':'Oprah','country':'Italy'}])
| "map" >> beam.Map(lambda x: (x['country']))
| "count" >> beam.combiners.Count.PerElement()
| "toDict" >> beam.combiners.ToDict()
| "print" >> beam.Map(print)
) 

p.run()

# Result {'USA': 2, 'Italy': 1}

【讨论】：

【解决方案2】：

这类似于字数统计示例。你可以在这里找到一个 python 实现 - https://beam.apache.org/get-started/wordcount-example/

【讨论】：