在 Apache Beam 中连接行

我无法理解 Apache Beam 中的连接（例如http://www.waitingforcode.com/apache-beam/joins-apache-beam/read http://www.waitingforcode.com/apache-beam/joins-apache-beam/read) 可以连接整行。

例如：

我有 2 个 CSV 格式的数据集，其中第一行是列标题。

首先：

a,b,c,d
1,2,3,4
5,6,7,8
1,2,5,4

第二：

c,d,e,f
3,4,9,10

我想在 c 列和 d 列上进行左连接，这样我最终会得到：

a,b,c,d,e,f
1,2,3,4,9,10
5,6,7,8,,
1,2,5,4,,

然而，Apache Beam 上的所有文档似乎都说 PCollection 对象需要是类型KV<K, V>加入时，所以我将我的 PCollection 对象分解为一个集合KV<String, String>对象（其中键是列标题，值是行值）。但在这种情况下（你只有一个带有值的键），我不知道如何维护行格式。 KV(c,7) 如何“知道”KV(a,5) 来自同一行？是Join https://beam.apache.org/documentation/sdks/javadoc/2.1.0/org/apache/beam/sdk/extensions/joinlibrary/Join.html到底是为了这种事情吗？

到目前为止我的代码：

PCollection<KV<String, String>> flightOutput = ...;
PCollection<KV<String, String>> arrivalWeatherDataForJoin = ...;
PCollection<KV<String, KV<String, String>>> output = Join.leftOuterJoin(flightOutput, arrivalWeatherDataForJoin, "");

Yes, Join是帮助像您这样的连接的实用程序类。它是一个包装器CoGropByKey, see 相应部分 https://beam.apache.org/documentation/programming-guide/#cogroupbykey在文档中。它的实现是很短 https://github.com/apache/beam/blob/master/sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java#L94. Its tests https://github.com/apache/beam/blob/master/sdks/java/extensions/join-library/src/test/java/org/apache/beam/sdk/extensions/joinlibrary/OuterLeftJoinTest.java#L38可能还有有用的例子。

您遇到的问题可能是由您选择密钥的方式引起的。

The KeyT int KV<KeyT,V1> in the Join库代表您用来匹配记录的键，它包含所有连接字段。因此，在您的情况下，您可能需要分配类似这样的键（伪代码）：

pCollection1:

    Key     Value
   (3,4)  (1,2,3,4)
   (7,8)  (5,6,7,8)
   (5,4)  (1,2,5,4)

pCollection2:

    Key     Value
   (3,4)  (3,4,9,10)

连接的结果将如下所示（伪代码）：

joinResultPCollection:

   Key              Value
  (3,4)      (1,2,3,4),(3,4,9,10)
  (7,8)      (5,6,7,8),nullValue
  (5,4)      (1,2,5,4),nullValue

因此，您可能需要在连接后添加另一个转换，以实际将左侧和右侧合并为组合行。

因为您有 CSV，所以您可能可以使用实际的字符串，例如"3,4"作为键（和值）。或者你可以使用Lists<>或您的自定义行类型。

例如，这正是Beam SQL Join 实现 https://github.com/apache/beam/blob/d110f6b7610b26edc1eb9a4b698840b21c151847/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamJoinRel.java#L250 does.