google-cloud-dataflow - 如何在beam sdk 2.0中创建自定义Combine.PerKey

Question

我们弄清楚了如何在beam sdk 2.0中创建自定义组合函数(经过大量猜测和beam sdk 2.0代码阅读)，因为数据流sdk 1.x语法在sdk 2.0中不起作用。

但是，我们无法弄清楚如何在beam sdk 2.0中创建自定义组合PER KEY功能。任何帮助或指示(或者更好的是一个实际的例子)将不胜感激。 (我们在互联网上搜索文档或示例，但没有找到；我们还尝试查看beam sdk 2.0的Combine类中的代码，但无法弄清楚，特别是因为PerKey类现在有一个私有(private)构造函数，所以我们不能再延长它了。)

如果有帮助的话，这里是我们如何在beam sdk 2.0中正确创建一个自定义组合器(没有)键，但我们不知道如何创建一个有> 一把 key :

public class CombineTemplateIntervalsIntoBlocks
        extends Combine.AccumulatingCombineFn<ImmutableMySetOfIntervals, TemplateIntervalAccum, ArrayList<ImmutableMySetOfIntervals>>{


    public CombineTemplateIntervalsIntoBlocks() {
    }

    @Override
    public TemplateIntervalAccum createAccumulator() {
        return new TemplateIntervalAccum()
    }

然后

public class TemplateIntervalAccum
        implements Combine.AccumulatingCombineFn.Accumulator<ImmutableMySetOfIntervals, TemplateIntervalAccum, ArrayList<ImmutableMySetOfIntervals>>, Serializable {
...

Answer 1

您不需要以不同的方式创建您的CombineFn 即可使用Combine.PerKey。

您可以扩展 AccumulatingCombineFn (将合并逻辑放入累加器中)或扩展 CombineFn (将合并逻辑放在 CombineFn 中)。还有其他选项，例如 BinaryCombineFn和IterableCombineFn .

假设您有 CombineFn<InputT, AccumT, OutputT>叫combineFn :

• 您可以使用Combine.globally(combineFn)创建 PTransform这需要 PCollection<InputT>并结合了所有元素。
• 或者，您可以使用Combine.perKey(combineFn)创建 PTransform这需要 PCollection<KV<K, InputT>>并组合与每个键关联的所有值并将它们组合起来。这对应于Combine.PerKey我相信您指的是。