采用 ndjson 格式的文本文件，以下代码生成我所期望的结果。一个 ndjson 文件，其中包含未嵌套的 quotes.USD 字典，并且删除了原始的 quotes 元素。
def unnest_quotes(element):
element['USDquotes'] = element['quotes']['USD']
del element['quotes']
return element
p = beam.Pipeline(options=pipeline_options)
ReadJson = p | ReadFromText(known_args.input,coder=JsonCoder())
MapFormattedJson = ReadJson | 'Map Function' >> beam.Map(unnest_quotes)
MapFormattedJson | 'Write Map Output' >> WriteToText(known_args.output,coder=JsonCoder())
但是，当我尝试使用 ParDo 实现相同的目的时，我不理解这种行为。
class UnnestQuotes(beam.DoFn):
def process(self,element):
element['USDquotes'] = element['quotes']['USD']
del element['quotes']
return element
p = beam.Pipeline(options=pipeline_options)
ReadJson = p | ReadFromText(known_args.input,coder=JsonCoder())
ClassFormattedJson = ReadJson | 'Pardo' >> beam.ParDo(UnnestQuotes())
ClassFormattedJson | 'Write Class Output' >> WriteToText(known_args.output,coder=JsonCoder())
这会生成一个文件，字典的每个键都在单独的一行上，没有值，如下所示。
"last_updated"
"name"
"symbol"
"rank"
"total_supply"
"max_supply"
"circulating_supply"
"website_slug"
"id"
"USDquotes"
就好像 Map 函数生成的 PCollection 是完整的字典，而 Pardo 为每个键生成一个 PCollection。
我知道我可以只使用 map 函数，但我需要了解这种行为，以便将来我确实需要使用 ParDo 时。
请您参考如下方法:
我在这个答案的帮助下解决了这个问题。 apache beam flatmap vs map
正如我所经历的，这与 FlatMap 和 Map 之间的区别是一样的。为了获得所需的行为，我需要做的就是将 Pardo 的返回值包装在一个列表中。
class UnnestQuotes(beam.DoFn):
def process(self,element):
element['USDquotes'] = element['quotes']['USD']
del element['quotes']
return [element]
