Scala API 扩展

    如果你想拥有完整的 Scala 体验，可以选择通过隐式转换增强 Scala API 的扩展。

    要使用所有可用的扩展，你只需为 DataStream API 添加一个简单的引入

    通常，DataStream API 不接受匿名模式匹配函数来解构元组、case 类或集合，如下所示：

    2. data.map {
    3.   case (id, name, temperature) => // [...]
    4.   // The previous line causes the following compilation error:
    5.   // "The argument types of an anonymous function must be fully known. (SLS 8.5)"
    6. }

    这个扩展在 DataStream Scala API 中引入了新的方法，这些方法在扩展 API 中具有一对一的对应关系。这些委托方法支持匿名模式匹配函数。

    DataStream API

    要单独使用此扩展，你可以添加以下引入：

    1. import org.apache.flink.api.scala.extensions.acceptPartialFunctions

    用于 DataSet 扩展

    1. object Main {
    2.   import org.apache.flink.streaming.api.scala.extensions._
    5.   def main(args: Array[String]): Unit = {
    6.     val env = StreamExecutionEnvironment.getExecutionEnvironment
    7.     val ds = env.fromElements(Point(1, 2), Point(3, 4), Point(5, 6))
    9.     ds.filterWith {
    10.       case Point(x, _) => x > 1
    11.     }.reduceWith {
    12.       case (Point(x1, y1), (Point(x2, y2))) => Point(x1 + y1, x2 + y2)
    13.     }.mapWith {
    14.       case Point(x, y) => (x, y)
    15.     }.flatMapWith {
    16.       case (x, y) => Seq("x" -> x, "y" -> y)
    17.     }.keyingBy {
    18.       case (id, value) => id
    19.     }
    20.   }