Apache Beam:FlatMap vs Map？

Question

我想了解在哪种情况下我应该使用FlatMap或Map.我的文档似乎并不清楚.

我仍然不明白在哪种情况下我应该使用FlatMap或Map的转换.

有人能给我一个例子,这样我才能理解他们的不同之处吗？

我理解Spark中FlatMap与Map的区别,但不确定是否有相似之处？

Answer 1

Beam中的这些变换与Spark(Scala也是如此)完全相同.

甲Map变换,映射从PCollectionN个元素的到另一个PCollectionN个元素的.

FlatMap变换将PCollectionsN个元素中的一个映射到零个或多个元素的N个集合中,然后将其展平为单个元素PCollection.

举个简单的例子,会发生以下情况:

beam.Create([1, 2, 3]) | beam.Map(lambda x: [x, 'any'])
# The result is a collection of THREE lists: [[1, 'any'], [2, 'any'], [3, 'any']]

鉴于:

beam.Create([1, 2, 3]) | beam.FlatMap(lambda x: [x, 'any'])
# The lists that are output by the lambda, are then flattened into a
# collection of SIX single elements: [1, 'any', 2, 'any', 3, 'any']

Answer 2

让我给你看一个例子

import apache_beam as beam

def categorize_explode(text):
  result = text.split(':')
  category = result[0]
  elements = result[1].split(',')
  return list(map(lambda x: (category, x), elements))

with beam.Pipeline() as pipeline:
  things = (
      pipeline
      | 'Categories and Elements' >> beam.Create(["Vehicles:Car,Jeep,Truck,BUS,AIRPLANE","FOOD:Burger,SANDWICH,ICECREAM,APPLE"])
      | 'Explode' >> beam.FlatMap(categorize_explode)
      | beam.Map(print)
  )

正如您所看到的，categorize_explode函数将字符串拆分为类别和相应的元素，并返回迭代器，例如[('Vehicles','Car'),('Vehicles','Jeep'),...]

FlatMap 获取此迭代器中的每个元素，并将每个元素视为 PCollection 中的单独元素。

所以结果是：

('Vehicles', 'Car')
('Vehicles', 'Jeep')
('Vehicles', 'Truck')
('Vehicles', 'BUS')
('Vehicles', 'AIRPLANE')
('FOOD', 'Burger')
('FOOD', 'SANDWICH')
('FOOD', 'ICECREAM')
('FOOD', 'APPLE')

而Map则执行一对一的映射。即该迭代器[('Vehicles','Car'),('Vehicles','Jeep'),...]将按原样返回。

所以 Map 的结果是：

[('Vehicles', 'Car'), ('Vehicles', 'Jeep'), ('Vehicles', 'Truck'), ('Vehicles', 'BUS'), ('Vehicles', 'AIRPLANE')]
[('FOOD', 'Burger'), ('FOOD', 'SANDWICH'), ('FOOD', 'ICECREAM'), ('FOOD', 'APPLE')]

我使用的方法有点类似于火花爆炸变换。

希望这可以帮助！！！