原文鏈接:https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/scala_api_extensions.html
為了保持Scala API和Java API之間相當(dāng)?shù)囊恢滦粤翰祝琒cala 的一些高階表達(dá)式被從標(biāo)準(zhǔn)的API中剔除檀何，包括批處理的和流處理的。
如果你想享受完整的Scala體驗(yàn)趁尼，你可以選擇通過隱式轉(zhuǎn)換來添加擴(kuò)展提升Scala的API埃碱。
為了使用所有的擴(kuò)展猖辫，你僅需要導(dǎo)入如下的包酥泞，在DataSet API中:

import org.apache.flink.api.scala.extensions._

或者在DataStream API中:

import org.apache.flink.streaming.api.scala.extensions._

或者你可以根據(jù)你自己的需要導(dǎo)入單獨(dú)的擴(kuò)展。
接受部分函數(shù)(Accept partial functions)
通常啃憎，DataSet API和DataStream API都不支持通過匿名模式匹配來構(gòu)建tuple芝囤、case class和集合，如下:

val data: DataSet[(Int, String, Double)] = // [...]
data.map {
  case (id, name, temperature) => // [...]
  // The previous line causes the following compilation error:
  // "The argument types of an anonymous function must be fully known. (SLS 8.5)"
}

這個(gè)擴(kuò)展為DataSet API和DataStream API引入了新的方法辛萍，這些方法與擴(kuò)展API有著一一對(duì)應(yīng)的關(guān)系悯姊，這些委派方法支持匿名模式匹配功能:
DataSet API
方法 原形 案例

mapWith map (DataSet)   data.mapWith {
  case (_, value) => value.toString
}
mapPartitionWith    mapPartition (DataSet)  data.mapPartitionWith {
  case head #:: _ => head
}
flatMapWith flatMap (DataSet)   data.flatMapWith {
  case (_, name, visitTimes) => visitTimes.map(name -> _)
}
filterWith  filter (DataSet)    data.filterWith {
  case Train(_, isOnTime) => isOnTime
}
reduceWith  reduce (DataSet, GroupedDataSet)    data.reduceWith {
  case ((_, amount1), (_, amount2)) => amount1 + amount2
}
reduceGroupWith reduceGroup (GroupedDataSet)    data.reduceGroupWith {
  case id #:: value #:: _ => id -> value
}
groupingBy  groupBy (DataSet)   data.groupingBy {
  case (id, _, _) => id
}
sortGroupWith   sortGroup (GroupedDataSet)  grouped.sortGroupWith(Order.ASCENDING) {
  case House(_, value) => value
}
combineGroupWith    combineGroup (GroupedDataSet)   grouped.combineGroupWith {
  case header #:: amounts => amounts.sum
}
projecting  apply (JoinDataSet, CrossDataSet)   data1.join(data2).
  whereClause(case (pk, _) => pk).
  isEqualTo(case (_, fk) => fk).
  projecting {
    case ((pk, tx), (products, fk)) => tx -> products
  }

data1.cross(data2).projecting {
  case ((a, _), (_, b) => a -> b
}
projecting  apply (CoGroupDataSet)  data1.coGroup(data2).
  whereClause(case (pk, _) => pk).
  isEqualTo(case (_, fk) => fk).
  projecting {
    case (head1 #:: _, head2 #:: _) => head1 -> head2
  }
}

DataStream API
方法 原形 案例

mapWith map (DataStream)    data.mapWith {
  case (_, value) => value.toString
}
mapPartitionWith    mapPartition (DataStream)   data.mapPartitionWith {
  case head #:: _ => head
}
flatMapWith flatMap (DataStream)    data.flatMapWith {
  case (_, name, visits) => visits.map(name -> _)
}
filterWith  filter (DataStream) data.filterWith {
  case Train(_, isOnTime) => isOnTime
}
keyingBy    keyBy (DataStream)  data.keyingBy {
  case (id, _, _) => id
}
mapWith map (ConnectedDataStream)   data.mapWith(
  map1 = case (_, value) => value.toString,
  map2 = case (_, _, value, _) => value + 1
)
flatMapWith flatMap (ConnectedDataStream)   data.flatMapWith(
  flatMap1 = case (_, json) => parse(json),
  flatMap2 = case (_, _, json, _) => parse(json)
)
keyingBy    keyBy (ConnectedDataStream) data.keyingBy(
  key1 = case (_, timestamp) => timestamp,
  key2 = case (id, _, _) => id
)
reduceWith  reduce (KeyedDataStream, WindowedDataStream)    data.reduceWith {
  case ((_, sum1), (_, sum2) => sum1 + sum2
}
foldWith    fold (KeyedDataStream, WindowedDataStream)  data.foldWith(User(bought = 0)) {
  case (User(b), (_, items)) => User(b + items.size)
}
applyWith   apply (WindowedDataStream)  data.applyWith(0)(
  foldFunction = case (sum, amount) => sum + amount
  windowFunction = case (k, w, sum) => // [...]
)
projecting  apply (JoinedDataStream)    data1.join(data2).
  whereClause(case (pk, _) => pk).
  isEqualTo(case (_, fk) => fk).
  projecting {
    case ((pk, tx), (products, fk)) => tx -> products
  }

想了解更多關(guān)于每個(gè)方法的語義信息，請(qǐng)參考DataSet和DataStream API文檔贩毕。
為了使用這些擴(kuò)展方法悯许，你需要引入如下類對(duì)于DataSet 來說:

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

對(duì)于DataStream來說:

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

下面的小例子中展示了如何組合使用這些擴(kuò)展的方法(使用DataSet API):

object Main {
  import org.apache.flink.api.scala.extensions._
  case class Point(x: Double, y: Double)
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val ds = env.fromElements(Point(1, 2), Point(3, 4), Point(5, 6))
    ds.filterWith {
      case Point(x, _) => x > 1
    }.reduceWith {
      case (Point(x1, y1), (Point(x2, y2))) => Point(x1 + y1, x2 + y2)
    }.mapWith {
      case Point(x, y) => (x, y)
    }.flatMapWith {
      case (x, y) => Seq("x" -> x, "y" -> y)
    }.groupingBy {
      case (id, value) => id
    }
  }
}