Scala API Extensions

为了保持 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 来使用那些你更喜欢的。

偏函数

通常，数据集和数据流 API 不接受匿名形式的函数去解构元组，例如类或集合，像下面这样：

val data: DataSet[(Int, String, Double)] = // [...]
data.map {
  case (id, name, temperature) => // [...]
  // 上面一行会引起下面的编译错误:
  // "匿名函数的参数类型必须完全可知. (SLS 8.5)"
}

这个扩展介绍了在数据集和数据流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 }`

获取更多方法的语法信息，请参考 DataSet 和 DataStream 的 API 帮助文档.

仅仅使用这一个扩展，你可以添加以下导入:

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

对数据集扩展导入

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

下面片段展示了如何用数据集 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
    }
  }
}