Scala API扩展
为了在Scala和Java API之间保持相当大的一致性，在批处理和流式传输的标准API中省略了一些允许Scala高级表达性的函数。
如果您想_享受完整的Scala体验_，可以选择选择关联通过隐式转换增强Scala API的扩展。
要使用所有可用的扩展，您只需import为DataSet API 添加一个简单的扩展
import org.apache.flink.api.scala.extensions._
或DataStream API
import org.apache.flink.streaming.api.scala.extensions._
或者，您可以导入单个扩展名_a-là-carte_以仅使用您喜欢的扩展名。
接受部分函数
通常，DataSet和DataStream API都不接受匿名模式匹配函数来解构元组，案例类或集合，如下所示：
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
方法：mapWith
原生：map（DataSet）
DEMO：
data.mapWith {
case (_, value) => value.toString
}
方法：mapPartitionWith
原生：mapPartition（DataSet）
DEMO：
data.mapPartitionWith {
case head #:: _ => head
}
方法：flatMapWith
原生：flatMap（DataSet）
DEMO：
data.flatMapWith {
case (_, name, visitTimes) => visitTimes.map(name -> _)
}
方法：filterWith
原生：filter（DataSet）
DEMO：
data.filterWith {
case Train(_, isOnTime) => isOnTime
}
方法：reduceWith
原生：reduce（DataSet，GroupedDataSet）
DEMO：
data.reduceWith {
case ((_, amount1), (_, amount2)) => amount1 + amount2
}
方法：reduceGroupWith
原生：reduceGroup（GroupedDataSet）
DEMO：
data.reduceGroupWith {
case id #:: value #:: _ => id -> value
}
方法：groupingBy
原生：groupBy（DataSet）
DEMO：
data.groupingBy {
case (id, _, _) => id
}
方法：sortGroupWith
原生：sortGroup（GroupedDataSet）
DEMO：
grouped.sortGroupWith(Order.ASCENDING) {
case House(_, value) => value
}
方法：combineGroupWith
原生：combineGroup（GroupedDataSet）
DEMO：
grouped.combineGroupWith {
case header #:: amounts => amounts.sum
}
方法：projecting
原生：apply（JoinDataSet，CrossDataSet）
DEMO：
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）
DEMO：
data1.coGroup(data2).
whereClause(case (pk, _) => pk).
isEqualTo(case (_, fk) => fk).
projecting {
case (head1 #:: _, head2 #:: _) => head1 -> head2
}
}
DataStream API
方法：mapWith
原生：map（DataStream）
DEMO：
data.mapWith {
case (_, value) => value.toString
}
方法：mapPartitionWith
原生：mapPartition（DataStream）
DEMO：
data.mapPartitionWith {
case head #:: _ => head
}
方法：flatMapWith
原生：flatMap（DataStream）
DEMO：
data.flatMapWith {
case (_, name, visits) => visits.map(name -> _)
}
方法：filterWith
原生：filter（DataStream）
DEMO：
data.filterWith {
case Train(_, isOnTime) => isOnTime
}
方法：keyingBy
原生：keyBy（DataStream）
DEMO：
data.keyingBy {
case (id, _, _) => id
}
方法：mapWith
原生：map（ConnectedDataStream）
DEMO：
data.mapWith(
map1 = case (_, value) => value.toString,
map2 = case (_, _, value, _) => value + 1
)
方法：flatMapWith
原生：flatMap（ConnectedDataStream）
DEMO：
data.flatMapWith(
flatMap1 = case (_, json) => parse(json),
flatMap2 = case (_, _, json, _) => parse(json)
)
方法：keyingBy
原生：keyBy（ConnectedDataStream）
DEMO：
data.keyingBy(
key1 = case (_, timestamp) => timestamp,
key2 = case (id, _, _) => id
)
方法：reduceWith
原生：reduce（KeyedStream，WindowedStream）
DEMO：
data.reduceWith {
case ((_, sum1), (_, sum2) => sum1 + sum2
}
方法：foldWith
原生：fold（KeyedStream，WindowedStream）
DEMO：
data.foldWith(User(bought = 0)) {
case (User(b), (_, items)) => User(b + items.size)
}
方法：applyWith
原生：apply（WindowedStream）
DEMO：
data.applyWith(0)(
foldFunction = case (sum, amount) => sum + amount
windowFunction = case (k, w, sum) => // [...] )
方法：projecting
原生：apply（JoinedStream）
DEMO：
data1.join(data2).
whereClause(case (pk, _) => pk).
isEqualTo(case (_, fk) => fk).
projecting {
case ((pk, tx), (products, fk)) => tx -> products
}
有关每种方法的语义的更多信息，请参阅 DataSet和DataStream 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
}
}
}
