Flink join上的丰富函数，Scala API

Question

我在Flink和Scala上很吃力。

我在DataSet上进行了一个连接转换，这个转换非常有效，但我想把它变成RichFuntion，这样我就可以访问广播集：

val newBoard: DataSet[Cell] = board.rightOuterJoin(neighbours)
                             .where("coords").equalTo("cellCoords"){
    (cell, neighbours) => {
            // Do some rich function things, like 
            // override the open method so I can get
            // the broadcasted set
    }
  }
}.withBroadcastSet(board, "aliveCells")

我已经查阅了所有的文档，但找不到任何在Scala中使用RichJoinFuntion的例子。我只找到map或filter中使用的丰富函数的示例，但join转换的语法不同（括号之间的函数与括号之间的功能）。

Answer 1

您可以将RichJoinFunction与Scala数据集API一起使用，如下所示

val newBoard: DataSet[Cell] = board.rightOuterJoin(neighbours)
                             .where("coords").equalTo("cellCoords")
                               .apply(new YourJoinFunction())
                               .withBroadcastSet(board, "aliveCells")
class YourJoinFunction extends RichJoinFunction[IN1, IN2, Cell] {
  override def join(first: IN1, second: IN2): Cell = {
    // Do some rich function things, like 
    // override the open method so I can get
    // the broadcasted set
  }
}