rakulang, dartlang, nimlang, golang, rustlang, lang lang no see

Scala API 扩展

焉知非鱼

Scala API Extensions

Scala API 扩展 #

为了在 Scala 和 Java API 之间保持相当程度的一致性,一些允许在 Scala 中进行高级表达的功能被从标准 API 中省略了,包括批处理和流式处理。

如果你想享受完整的 Scala 体验,你可以选择加入通过隐式转换来增强 Scala API 的扩展。

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

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

或者 DataStream API:

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

另外,你也可以按顺序导入单个扩展,只使用你喜欢的扩展。

接受部分函数 #

通常情况下,DataSet 和 DataStream API 都不接受匿名模式匹配函数来解构 tuple、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 #

  • 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 #

  • mapWith 方法和原来的 map (DataStream)
data.mapWith {
  case (_, value) => value.toString
}
  • 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 (KeyedStream, WindowedStream)
data.reduceWith {
  case ((_, sum1), (_, sum2) => sum1 + sum2
}
  • foldWith 方法和原来的 fold (KeyedStream, WindowedStream)
data.foldWith(User(bought = 0)) {
  case (User(b), (_, items)) => User(b + items.size)
}
  • applyWith 方法和原来的 apply (WindowedStream)
data.applyWith(0)(
  foldFunction = case (sum, amount) => sum + amount
  windowFunction = case (k, w, sum) => // [...]
)
  • projecting 方法和原来的 apply (JoinedStream)
data1.join(data2).
  whereClause(case (pk, _) => pk).
  isEqualTo(case (_, fk) => fk).
  projecting {
    case ((pk, tx), (products, fk)) => tx -> products
  }

关于每个方法的语义的更多信息,请参考 DataSetDataStream API 文档。

要专门使用这个扩展,可以添加以下导入。

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

原文链接: https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/scala_api_extensions.html