为了保存Scala和Java API之间的一致性，一些允许Scala使用高层次表达式的特性从批处理和流处理的标准API中删除。
如果你想体验Scala表达式的全部特性，你可以通过隐式转换(implicit conversions)来加强Scala API。
为了使用这些扩展，在DataSet API中，你仅仅需要引入下面类：
import org.apache.flink.api.scala.extensions._
而在 DataStream API中，你需要引入下面类：
import org.apache.flink.streaming.api.scala.extensions._
当然，你也可以引入一个具体的类，并仅仅使用这个类中的特性。
使用偏函数(partial functions)
通常情况下，DataSet和DataStream APIs都不支持匿名模式匹配函数来解构tuples, case classes或者collections，如下：
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都提供了相应的函数扩展，并提供了对匿名模式匹配函数的支持。具体的函数及其使用如下：
DataSet API
| Method | Original | Example |
|---|---|---|
| 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
| Method | Original | Example |
|---|---|---|
| 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
}
|
更多关于上面每个函数的语言，可以分别参见DataStream和DataSet API。
如果想精确地使用其中一个函数，在DataSet中可以如下使用：
import org.apache.flink.api.scala.extensions.acceptPartialFunctions
而在DataStream中可以如下使用：
import org.apache.flink.streaming.api.scala.extensions.acceptPartialFunctions
下面的代码片段展示如何在DataSet中使用这些扩展：
object Iteblog {
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
}
}
}
本文翻译自：https://ci.apache.org/projects/flink/flink-docs-master/apis/scala_api_extensions.html
本博客文章除特别声明，全部都是原创！原创文章版权归过往记忆大数据（过往记忆）所有，未经许可不得转载。
本文链接: 【Flink：Scala API函数扩展】（https://www.iteblog.com/archives/1651.html）
