Apache Flink中的Join输出

Question

在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个数据集的连接？这将使示例源更加混乱.

谢谢你的帮助.

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的优化器可以避免进行不必要的改组.转发字段注释告诉优化器,某些字段(例如连接键)未被连接函数修改,并允许重用现有的分区和排序.