使用 SideInput 加入集合答案

【问题标题】：Join the collection using SideInput使用 SideInput 加入集合
【发布时间】：2019-07-26 02:10:52
【问题描述】：

尝试使用 SideInput 转换加入两个 Pcollection。在 ParDo 函数中映射值时，我们可以从 sideinput 集合中获取多个映射记录作为一个集合。在这种情况下，如何处理集合以及如何将这些值集合返回到 PCollection。

如果有人能帮助解决这个问题，那就太好了。这是我尝试过的代码sn-p。

PCollection<TableRow> pc1 = ...;
PCollection<Row> pc1Rows = pc1.apply(
    ParDo.of(new fnConvertTableRowToRow())).setRowSchema(schemaPc1);
PCollection<KV<Integer, Row>> keyed_pc1Rows = pc1Rows.apply(
    WithKeys.of(new SerializableFunction<Row, Integer>() {
       public Integer apply(Row s) {
         return Integer.parseInt(s.getValue("LOCATION_ID").toString());
       }
    }));

PCollection<TableRow> pc2 = ...;

PCollection<Row> pc2Rows = pc2.apply(
    ParDo.of(new fnConvertTableRowToRow())).setRowSchema(schemaPc2);

PCollection<KV<Integer, Iterable<Row>>> keywordGroups = pc2Rows.apply(
    new fnGroupKeyWords());

PCollectionView<Map<Integer, Iterable<Row>>> sideInputView =
    keywordGroups.apply("Side Input",
       View.<Integer, Iterable<Row>>asMap());

PCollection<Row> finalResultCollection = keyed_pc1Rows.apply("Process",
  ParDo.of(new DoFn<KV<Integer,Row>, Row>() {

    @ProcessElement
    public void processElement(ProcessContext c) {

      Integer key = Integer.parseInt(c.element().getKey().toString());

      Row leftRow = c.element().getValue();

      Map<Integer, Iterable<Row>> key2Rows = c.sideInput(sideInputView);

      Iterable<Row> rightRowsIterable = key2Rows.get(key);

      for (Iterator<Row> i = rightRowsIterable.iterator(); i.hasNext(); ) {
        Row suit = (Row) i.next();
        Row targetRow = Row.withSchema(schemaOutput)
                           .addValues(leftRow.getValues())
                           .addValues(suit.getValues())
                           .build();
        c.output(targetRow);
      }
    }
}).withSideInputs(sideInputView));

public static class fnGroupKeyWords extends
  PTransform<PCollection<Row>, PCollection<KV<Integer, Iterable<Row>>>> {

  @Override
  public PCollection<KV<Integer, Iterable<Row>>> expand(
    PCollection<Row> rows) {

      PCollection<KV<Integer, Row>> kvs = rows.apply(
          ParDo.of(new TransferKeyValueFn()));
      PCollection<KV<Integer, Iterable<Row>>> group = kvs.apply(
          GroupByKey.<Integer, Row> create());
      return group;
  }
}


public static class TransferKeyValueFn extends
  DoFn<Row, KV<Integer, Row>> {

  @ProcessElement
  public void processElement(ProcessContext c) throws ParseException {
    Row tRow = c.element();

    c.output(
       KV.of(
          Integer.parseInt(tRow.getValue("DW_LOCATION_ID").toString()),
          tRow));
  }
}

【问题讨论】：

标签： google-cloud-dataflow apache-beam

【解决方案1】：

如果您希望使用一个公共密钥将两个 PCollection 连接在一起。 CoGroupByKey 可能更有意义。请考虑这种方法而不是侧面输入

这个blog post也有很好的解释。

【讨论】：

感谢您的反馈。我们已经尝试过使用 CoGroupsByKey 并在处理超过 10000 个元素时遇到了该方法的性能问题。根据 stackoverflow 的建议 link，使用 SideInput 进行连接。现在我们面临着 SideInput 的挑战，当集合有多个主要数据的映射记录并且想知道如何处理它时。

【解决方案2】：

我认为，如果您有一个可以放入内存的非常小的集合，那么使用 SideInput 建议会很好。您可以将其用作view.asMultimap 的侧面输入。然后在 ParDo 处理更大的 PCollection （在 GBK 之后，为您提供对所有元素的可迭代键），从侧面输入中查找您感兴趣的键。这是一个使用多映射 pcollection 的example test pipeline。

但是，如果您的集合非常大，那么使用 Flatten 将两个 pcollections 组合在一起会是一种更好的方法。然后使用 GroupByKey ，这将为您提供同一键下的元素的可迭代。这仍将按顺序处理。不过，我相信您会遇到性能问题，除非您消除热键。 Please see the explanation of using combiners to alleviate this.

【讨论】：