Scala API 扩展#

为了在 Scala 和 Java API 之间保持一定的一致性,批处理和流处理的标准 API 中保留了一些Scala的高级表达的特性。

如果想享受完整的Scala体验,您可以选择通过隐式转换增强 Scala API 的功能。

要使用所有可用的扩展,您只需为 DataSet API 添加一个简单的导入语句

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

或 DataStream API

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

或者,您也可以导入扩展 a-là-carte,以便只使用您喜欢的extensions。

Accept partial functions#

通常,DataSet 和 DataStream API 都不接受以匿名模式匹配函数来解构元组,case 类或集合,如下:

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)"
}

这个扩展引入了 DataSet 和 DataStream Scala API 中的新方法,它们在扩展 API 中具有一对一的对应关系。这些委托方法支持匿名模式匹配功能。

DataSet API#

Method Original Example
mapWith map (DataSet) ```scala data.mapWith { case (_, value) => value.toString } ```
mapPartitionWith mapPartition (DataSet) ```scala data.mapPartitionWith { case head #:: _ => head } ```
flatMapWith flatMap (DataSet) ```scala data.flatMapWith { case (_, name, visitTimes) => visitTimes.map(name -> _) } ```
filterWith filter (DataSet) ```scala data.filterWith { case Train(_, isOnTime) => isOnTime } ```
reduceWith reduce (DataSet, GroupedDataSet) ```scala data.reduceWith { case ((_, amount1), (_, amount2)) => amount1 + amount2 } ```
reduceGroupWith reduceGroup (GroupedDataSet) ```scala data.reduceGroupWith { case id #:: value #:: _ => id -> value } ```
groupingBy groupBy (DataSet) ```scala data.groupingBy { case (id, _, _) => id } ```
sortGroupWith sortGroup (GroupedDataSet) ```scala grouped.sortGroupWith(Order.ASCENDING) { case House(_, value) => value } ```
combineGroupWith combineGroup (GroupedDataSet) ```scala grouped.combineGroupWith { case header #:: amounts => amounts.sum } ```
projecting apply (JoinDataSet, CrossDataSet) ```scala 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 }

      </td>
    </tr>
    <tr>
      <td><strong>projecting</strong></td>
      <td><strong>apply (CoGroupDataSet)</strong></td>
      <td>
```scala
data1.coGroup(data2).
  whereClause(case (pk, _) => pk).
  isEqualTo(case (_, fk) => fk).
  projecting {
    case (head1 #:: _, head2 #:: _) => head1 -> head2
  }
}
  </td>
</tr>
</tr>

DataStream API#

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

有关每种方法的更多语义信息,请参阅 [DataSet]({{ site.baseurl }}/dev/batch/index.html) 和 [DataStream]({{ site.baseurl }}/dev/datastream_api.html) 的API 文档.

您可以添加以下的 import 语句来专门使用此扩展:

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

对于 DataSet 扩展

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

以下的代码片段展示了如何一起使用这些扩展方法(使用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
    }
  }
}