加入两个大容量的 PCollection 有性能问题

Question

将两个 Pcollection 与 CoGroupsByKey 方法结合起来需要数小时才能执行 8 多万条记录。从另一个 stackoverflow 帖子 CoGbkResult has more than 10000 elements,reiteration (which may be slow) is required 中注意到“CoGbkResult 有超过 10000 个元素，需要重复（可能很慢）。”

使用这种方法提高性能的任何建议。

这里是代码sn-p，

PCollection<TableRow> pc1 = ...;
PCollection<TableRow> pc2 = ...;

WithKeys<String, TableRow> withKeyValue = 
  WithKeys.of((TableRow row) -> String.format("%s",row.get("KEYNAME")))
          .withKeyType(TypeDescriptors.strings());

PCollection<KV<String,TableRow>> keyed_pc1 =
  pc1.apply("WithKeys", withKeyValue );

PCollection<KV<String,TableRow>> keyed_pc2 = 
  pc2.apply("WithKeys", withKeyValue );

// (org.apache.beam.sdk.extensions.joinlibrary.Join class)
PCollection<KV<String,KV<TableRow,TableRow>>> joinedCollection = 
  Join.innerJoin(keyed_pc1, keyed_pc2); 

Answer 1

Apache Beam 规范没有定义连接的执行，除了 SDK 之外，没有更快的方法来编写内部连接。因此，这个问题的答案取决于执行连接的内容，即哪个运行者。我不知道 Flink 或 Spark 跑步者，所以这个答案将特定于 Dataflow 跑步者。

如果您还没有，请查看topic 上的这篇博文。在博文中，它描述了可以手动启用的 Dataflow Shuffle Service。此服务比当前的默认实现更好，通常可以更快地执行，尤其是对于连接。

要启用Dataflow Shuffle Service，请传入以下flags：

--experiments=shuffle_mode=service
--region=<allowed region>

允许随机播放的区域为：“us-central1”、“europe-west1”、“europe-west4”、“asia-northeast1”。

Answer 2

我的理解是你的加入有一个热键：一个有很多条目的键，结果条目不适合工作人员的内存。这意味着当您稍后使用它时，可能会导致重新获取可能会降低性能的数据。

Join.innerJoin 仍然使用CoGBK internally 进行连接，因此仅使用该库不一定更有效。不过，迭代集合的顺序可能很重要。

如果您在一侧有小的 pcollection（适合内存），您可以使用查找表的方法进行连接。参考JoinAsLookup。

如果你有办法知道哪个键是热的，你可以在加入之前把它分成更小的键，但这需要在工程方面做更多的工作和对数据有一些预知。