0

我有一个 PCollection ，其中每个元素都是一个键，值元组是这样的：(key, (value1,..,value_n) )

我需要将此 PCollection 拆分为两个处理分支。

与往常一样，我需要整个管道尽可能快并使用尽可能少的内存。

我想到了两个想法：

选项 1：使用具有多个输出的 DoFn 拆分 PColl

class SplitInTwo(beam.DoFn):

   def process(self, kvpair):
       key, values = kvpair
       
       yield beam.TaggedOutput('left', (key, values[0:2]))
       yield beam.TaggedOutput('right', (key, values[2:]))

class ProcessLeft(beam.DoFn):
   def process(self, kvpair):
       key,values = kvpair
       ...
       yield (key, results)

# class ProcessRight is similar to ProcessLeft

然后像这样构建管道

   splitme = pcoll | beam.ParDo(SplitInTwo()).with_outputs('left','right')
   left = splitme.left | beam.ParDo(ProcessLeft())
   right = splitme.right | beam.ParDo(ProcessRight())

选项 2：在原始 PCollection 上使用两个不同的 DoFn

另一种选择是使用两个 DoFns 来读取和处理同一个 PCollection。仅将一个用于数据的“左侧”和“右侧”：

class ProcessLeft(beam.DoFn):

   def process(self, kvpair):
       key = kvpair[0]
       values = kvpair[0][0:2]
       ...
       yield (key,result)

# class ProcessRight is similar to ProcessLeft

构建管道更简单......（而且您不需要跟踪您拥有哪些标记输出）：

   left = pcoll | beam.ParDo(ProcessLeft())
   right = pcoll| beam.ParDo(ProcessRight())

但是……更快吗？将需要比第一个更少的内存？

（我正在考虑第一个选项可能会被跑步者融合 - 而不仅仅是数据流跑步者）。

4

1 回答 1

2

在这种情况下，两个选项都将由 runner 融合，因此两个选项在性能方面会有些相似。如果您想将数据重新洗牌到单独的工作人员中，那么选项 1是您的最佳选择，因为读取的序列化集合ProcessLeft会ProcessRight更小。

   splitme = pcoll | beam.ParDo(SplitInTwo()).with_outputs('left','right')
   left = splitme.left | beam.Reshuffle() | beam.ParDo(ProcessLeft())
   right = splitme.right | beam.Reshuffle() | beam.ParDo(ProcessRight())

转换将Reshuffle确保您的数据被写入中间洗牌，然后在下游消费。这会破坏融合。

于 2020-12-04T18:13:43.853 回答