在 Apache Beam 中加入行

Question

我无法理解 Apache Beam 中的连接（例如 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 是不是专门针对这种事情的？

到目前为止我的代码：

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

Answer 1

是的，Join 是实用程序类，可帮助您处理像您这样的连接。它是CoGropByKey 的包装，请参阅文档中的the corresponding section。它的实现是pretty short。它的tests 也可能有有用的例子。

您的问题可能是由您选择键的方式引起的。

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

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

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

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

例如，这正是Beam SQL Join implementation 所做的。