Apache Flink 中 Join 的输出答案

【问题标题】：Output of Join in Apache FlinkApache Flink 中 Join 的输出
【发布时间】：2015-06-11 15:55:53
【问题描述】：

在 Apache Flink 中，如果我在一个主键上连接两个数据集，我会得到一个元组 2，其中包含每个数据集对应的数据集条目。

问题是，当将 map() 方法应用于输出的元组 2 数据集时，它看起来并不好，尤其是如果两个数据集的条目都具有大量特征。

在两个输入数据集中使用元组可以得到如下代码：

var in1: DataSet[(Int, Int, Int, Int, Int)] = /* */
var in2: DataSet[(Int, Int, Int, Int)] = /* */

val out = in1.join(in2).where(0, 1, 2).equalTo(0, 1, 2)
  .map(join => (join._1._1, join._1._2, join._1._3,
                    join._1._4, join._1._5, join._2._4))

我不介意使用 POJO 或案例类，但我不知道这会如何使它变得更好。

问题 1： 有没有一种很好的方法来展平元组 2？例如。使用另一个运算符。

问题 2： 如何处理 3 个数据集在同一个键上的连接？它会使示例源更加混乱。

感谢您的帮助。

【问题讨论】：

标签： scala apache-flink

【解决方案1】：

你可以直接在每对被连接的元素上应用一个连接函数，例如

val leftData: DataSet[(String, Int, Int)] = ...
val rightData: DataSet[(String, Int)] = ...
val joined: DataSet[(String, Int, Int)] = leftData
      .join(rightData).where(0).equalTo(0) { (l, r) => (l._1, l._2, l._3 + r._2) ) }

为了回答第二个问题，Flink 只处理二元连接。但是，如果您提供有关函数行为的提示，Flink 的优化器可以避免进行不必要的洗牌。 Forward Field annotations 告诉优化器，某些字段（例如连接键）没有被您的连接函数修改，并且可以重用现有的分区和排序。

【讨论】：