Apache Flink中Join的输出

Question

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

问题是，当将map（）方法应用于即将到来的tuple 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或case类，但我不明白这会如何使它变得更好。

问题1:有没有一个很好的方法来扁平化元组2？例如使用另一个运营商。

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

谢谢你的帮助。

Answer 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 批注告诉优化程序，某些字段（如连接键）尚未被连接函数修改，并允许重用现有的分区和排序。