Scala API扩展

为了在Scala和Java API之间保持相当大的一致性，在批处理和流式传输的标准API中省略了一些允许Scala高级表达性的函数。

如果您想享受完整的Scala体验，可以选择选择关联通过隐式转换增强Scala API的扩展。

要使用所有可用的扩展，您只需import为DataSet API 添加一个简单的扩展

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

或DataStream API

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

或者，您可以导入单个扩展名a-là-carte以仅使用您喜欢的扩展名。

接受部分函数

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

  1. val data: DataSet[(Int, String, Double)] = // [...]
  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. }

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

DataSet API

方法原生DEMO
mapWithmap（DataSet）
  1. data.mapWith {  case (, value) => value.toString}
mapPartitionWithmapPartition（DataSet）
  1. data.mapPartitionWith {  case head #::  => head}
flatMapWithflatMap（DataSet）
  1. data.flatMapWith {  case (, name, visitTimes) => visitTimes.map(name -> )}
filterWithfilter（DataSet）
  1. data.filterWith {  case Train(, isOnTime) => isOnTime}
reduceWithreduce（DataSet，GroupedDataSet）
  1. data.reduceWith {  case ((, amount1), (, amount2)) => amount1 + amount2}
reduceGroupWithreduceGroup（GroupedDataSet）
  1. data.reduceGroupWith {  case id #:: value #::  => id -> value}
groupingBygroupBy（DataSet）
  1. data.groupingBy {  case (id, , ) => id}
sortGroupWithsortGroup（GroupedDataSet）
  1. grouped.sortGroupWith(Order.ASCENDING) {  case House(, value) => value}
combineGroupWithcombineGroup（GroupedDataSet）
  1. grouped.combineGroupWith {  case header #:: amounts => amounts.sum}
projectingapply（JoinDataSet，CrossDataSet）
  1. 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}
projectingapply（CoGroupDataSet）
  1. data1.coGroup(data2).  whereClause(case (pk, ) => pk).  isEqualTo(case (, fk) => fk).  projecting {    case (head1 #:: , head2 #:: _) => head1 -> head2  }}

DataStream API

方法原生DEMO
mapWithmap（DataStream）
  1. data.mapWith {  case (, value) => value.toString}
mapPartitionWithmapPartition（DataStream）
  1. data.mapPartitionWith {  case head #::  => head}
flatMapWithflatMap（DataStream）
  1. data.flatMapWith {  case (, name, visits) => visits.map(name -> )}
filterWithfilter（DataStream）
  1. data.filterWith {  case Train(, isOnTime) => isOnTime}
keyingBykeyBy（DataStream）
  1. data.keyingBy {  case (id, , ) => id}
mapWithmap（ConnectedDataStream）
  1. data.mapWith(  map1 = case (, value) => value.toString,  map2 = case (, , value, ) => value + 1)
flatMapWithflatMap（ConnectedDataStream）
  1. data.flatMapWith(  flatMap1 = case (, json) => parse(json),  flatMap2 = case (, , json, ) => parse(json))
keyingBykeyBy（ConnectedDataStream）
  1. data.keyingBy(  key1 = case (, timestamp) => timestamp,  key2 = case (id, , ) => id)
reduceWithreduce（KeyedStream，WindowedStream）
  1. data.reduceWith {  case ((, sum1), (, sum2) => sum1 + sum2}
foldWithfold（KeyedStream，WindowedStream）
  1. data.foldWith(User(bought = 0)) {  case (User(b), (, items)) => User(b + items.size)}
applyWithapply（WindowedStream）
  1. data.applyWith(0)(  foldFunction = case (sum, amount) => sum + amount  windowFunction = case (k, w, sum) => // […])
projectingapply（JoinedStream）
  1. data1.join(data2).  whereClause(case (pk, ) => pk).  isEqualTo(case (_, fk) => fk).  projecting {    case ((pk, tx), (products, fk)) => tx -> products  }

有关每种方法的语义的更多信息，请参阅DataSet和DataStream API文档。

要仅使用此扩展程序，您可以添加以下内容import：

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

对于DataSet扩展和

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

以下代码段显示了如何一起使用这些扩展方法的最小示例（使用DataSet API）：

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