Question

为了在 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
  }
  }
}