评估整个 PCollection 的函数答案

【问题标题】：Evaluate function on entire PCollection评估整个 PCollection 的函数
【发布时间】：2018-03-12 07:03:05
【问题描述】：

我有一个PCollection<V>，我想从本质上评估一个函数f(List<V>)。我可以制作一个CombineFn，看起来像：

class GlobalCombineFn extends Combine.CombineFn<Write, List<Write>, Void> {
    @Override
    public List<Write> createAccumulator() {
        return new ArrayList<>();
    }

    @Override
    public List<Write> addInput(List<Write> accumulator, Write input) {
        accumulator.add(input);
        return accumulator;
    }

    @Override
    public List<Write> mergeAccumulators(Iterable<List<Write>> accumulators) {
        List<Write> result = createAccumulator();
        accumulators.forEach(result::addAll);
        return result;
    }

    @Override
    public Void extractOutput(List<Write> accumulator) {
        f(accumulator);
        return null;
    }
}

但是，这有点愚蠢，而且有很多样板。有没有内置的方法可以做到这一点？

我也尝试过使用View.asIterable()/View.asList()，但是没有任何方法可以对PCollectionView进行操作。

【问题讨论】：

标签： apache-beam

【解决方案1】：

我最终只是为此编写了自己的 PTransform，这要简单得多：

public class ReduceToIterable<T> extends PTransform<PCollection<T>, PCollection<Iterable<T>>> {
    @Override
    public PCollection<Iterable<T>> expand(PCollection<T> input) {
        return input.apply(WithKeys.of((Void) null))
                .setCoder(KvCoder.of(VoidCoder.of(), input.getCoder()))
                .apply(GroupByKey.create())
                .apply(Values.create());
    }
}

【讨论】：