在累积来自所有节点的数据后，使用CombineFn组合每个键的所有值

Question

我想在perKey的基础上迭代KV pCollection的值。我使用下面的代码组合使用自定义类，

PCollection<KV<String, String>> combinesAttributes =
              valExtract.get(extAttUsers).apply(Combine.<String, String>perKey(
                      new CombineAttributes()));

下面是我的自定义组合类，

public static class CombineAttributes implements SerializableFunction<Iterable<String>, String> {
   @Override
   public String apply(Iterable<String> input) {...}..}

incompatible types: CombineAttributes cannot be converted to com.google.cloud.dataflow.sdk.transforms.SerializableFunction<java.lang.Iterable<java.lang.String>,java.lang.String>

组合函数

public static class CombineAttributes extends CombineFn<Iterable<String>, CombineAttributes.Accum, String> {

public static class Accum {
  List<String> inputList = new ArrayList<String>();
}
public Accum createAccumulator() { return new Accum(); }
public Accum addInput(Accum accum, Iterable<String> input) {
  for (String item : input) {
    accum.inputList.add(item);
  }
  return accum;
}
public Accum mergeAccumulators(Iterable<Accum> accums) {
   Accum merged = createAccumulator();
   for (Accum accum : accums) {
     for (String item : accum.inputList) {
       merged.inputList.add(item);
     }
   }
   return merged;
 }
 public String extractOutput(Accum accum) {
   return "";
 }
}

没有可用于合并perKey扩展combineFN的示例代码。请让我知道上面的代码有什么问题。

Answer 1

combineAttributes（无论以何种方式实现）将首先在每台机器上调用。因此，它可以将[attr1a,attr1b]组合成单个字符串或累加器（例如attr1a+attr1b)。然后，它将在另一台机器上运行，以将[attr1c]合并为attr1c。然后，它将合并所有这些部分结果，得到最终的累加器--attr1a+attr1b+attr1c。在原始实现的情况下，这将是最终的答案。在后者中，将对该累加器调用extractoutput。