Flink DataStream API之Operators

官网位置：https://ci.apache.org/projects/flink/flink-docs-release-1.9/zh/dev/stream/operators/

Operators transform one or more DataStreams into a new DataStream. Programs can combine multiple transformations into sophisticated dataflow topologies.

算子可以将一个或多个DataStream转化为新的DataStream。程序可以将多种转换组合成复杂的数据流拓扑。下面将介绍一下Flink DataStream API中一些常用的算子，注意：本文是以Scala代码为例，有些语法在Java里不通用。

1 DataStream Transformations

官网以表格的形式列了一些转换类的算子，有些可能不好理解，这里以实际的代码例子进行演示。

DataStream常用转换如下图所示：

1.1 map

类型转换：

DataStream->DataStream

描述：

熟悉scala的同学，对于这个map应该很好理解，map操作是依次取DataStream中一个元素并生成另一个元素。

调用方式:

1. 传入一个继承自MapFunction[T,R]的实例

def map[R: TypeInformation](mapper: MapFunction[T, R]): DataStream[R]

2. 函数式调用，传入一个参数类型为T，返回值为R的函数

  def map[R: TypeInformation](fun: T => R): DataStream[R] 

代码演示：

先通过1到10的集合生成一个DataStream，并通过map操作，将其每个元素都乘以2，生成新的DataStream。

    // create DataStream from collection
    val data = env.fromCollection(1 to 10)

    // 将每个元素*2
    data.map(_ * 2).print("map")

输出：

map:7> 10
map:4> 4
map:3> 2
map:2> 16
map:8> 12
map:6> 8
map:5> 6
map:1> 14
map:3> 18
map:4> 20

1.2 flatMap

类型转换：

DataStream->DataStream

描述：

flatMap与map类似，只是它是依次取DataStream中的一个元素生成0个或多个元素。

调用方式：

它在scala里提供了三种调用方式：

1. 传入一个继承自FlatMapFunction[T, R]的类实例

def flatMap[R: TypeInformation](flatMapper: FlatMapFunction[T, R]): DataStream[R]

2. 传入参数为T和Collector[R]无返回值的函数

 def flatMap[R: TypeInformation](fun: (T, Collector[R]) => Unit): DataStream[R]

3. 传入参数类型为T，返回值类型为TraversableOnce(集合特质，常见集合都继承该特质)的函数

def flatMap[R: TypeInformation](fun: T => TraversableOnce[R]): DataStream[R]

代码演示：

先通过Array(1,2,3)和Array(4,5,6)两个元素生成一个DataStream，并通过flatMap操作，将两个数组展开

    // create DataStream from elements
    val data = env.fromElements(Array(1,2,3),Array(4,5,6))
    // 将数据展开
    data.flatMap(_.toList).print("flatMap")

结果：

flatMap:1> 4
flatMap:8> 1
flatMap:1> 5
flatMap:8> 2
flatMap:1> 6
flatMap:8> 3

1.3 filter

类型转换：

DataStream->DataStream

描述：

按照条件过滤，得到满足条件的元素生成新的DataStream

调用方式：

调用方式有两种：

1. 传入一个继承自FilterFunction[T]类实例

def filter(filter: FilterFunction[T]): DataStream[T]

2. 传入一个入参类型为T，返回值类型为Boolean的函数

def filter(fun: T => Boolean): DataStream[T]

代码演示：

先通过1 到10的集合生成一个DataStream，并通过filter操作，过滤出奇数

    // create DataStream from collection
    val data = env.fromCollection(1 to 10)
    // 输出奇数
    data.filter(_ % 2 == 1).print("filter")

输出：

filter:3> 1
filter:5> 3
filter:7> 5
filter:1> 7
filter:3> 9

1.4 keyBy

类型转换：

DataStream->KeyedStream

描述：

根据指定的Key选择器，进行分区，生成KeyedStream

调用方式：

在scala里key选择器有四种指定方式：

1. 对于tuple和array类型，我们可以指定一个或多个key的位置

def keyBy(fields: Int*): KeyedStream[T, JavaTuple]

2. 指定一个或多个key的字段表达式

def keyBy(firstField: String, otherFields: String*): KeyedStream[T, JavaTuple]

3. 传入key选择函数

def keyBy[K: TypeInformation](fun: T => K): KeyedStream[T, K]

4. 实现key选择器KeySelector，并传入该实例

def keyBy[K: TypeInformation](fun: KeySelector[T, K]): KeyedStream[T, K]

代码演示：

先通过集合生成一个DataStream，然后根据第一个元素进行KeyBy

    // create DataStream from collection
    val data = env.fromCollection(List(("device-1", 1), ("device-2", 2), ("device-1", 2)))
    data.keyBy(_._1)

输出：

keyed:3> (device-2,2)
keyed:1> (device-1,1)
keyed:1> (device-1,2)

1.5 reduce

类型转换：

KeyedStream->DataStream

描述：

在KeyedStream上，将当前值以及上次计算结果应用到指定的reduce函数中得到新的值，依次计算

调用方式：

在scala里reduce调用有两种方式：

1. 实现ReduceFunction[T]接口中的reduce方法，并传入该实例

 def reduce(reducer: ReduceFunction[T]): DataStream[T]

2. 传入参数类型为T,T返回值为T的函数

def reduce(fun: (T, T) => T): DataStream[T] 

代码演示：

将上面KeyBy之后生成的KeyedStream，应用reduce算子，进行累加

    keyedStream.reduce((value1, value2) => {
      (value1._1, value1._2 + value2._2)
    }).print("reduce")

输出：

reduce:1> (device-1,1)
reduce:3> (device-2,2)
reduce:1> (device-1,3)

1.6 fold(deprecated)

类型转换：

KeyedStream->DataStream

描述：

在KeyedStream上，与reduce类似，只是多了一个初始值。

调用方式：

在scala里fold调用有两种方式：

1. 实现FoldFunction[T,R]接口中的fold方法，并传入该实例,另外还需要传入初始化值。

  def fold[R: TypeInformation](initialValue: R, folder: FoldFunction[T,R]): 
      DataStream[R]

2. 柯里化传参，第一个参数为初始值，第二个参数为函数表达式

def fold[R: TypeInformation](initialValue: R)(fun: (R,T) => R): DataStream[R]

代码演示：

将上面KeyBy之后生成的KeyedStream，应用fold算子，进行字符串拼接

    keyedStream.fold("start")((accumulator, value) => {
      s"$accumulator-${value._1}-${value._2}"
    }).print("fold")

输出：

fold:1> start-device-1-1
fold:3> start-device-2-2
fold:1> start-device-1-1-device-1-2

1.7 aggregations

类型转换：

KeyedStream->DataStream

描述：

在KeyedStream上，应用聚合方法，Flink提供内置的聚合方法有：sum、min、minBy、maxBy。这里解释一下min和minBy，max和maxBy之间的却别。以max和maxBy为例，max(position|field)只是返回指定的position或field的最大值，其它字段的值为第一次出现的记录值，而maxBy(position|field)则是返回position或field最大值所在的记录。怎么理解呢？假如streamData:DataStream[DeviceEvent]有两条记录 DeviceEvent(1,1,1)、 DeviceEvent(2,2,2)，即

val streamData:DataStream[DeviceEvent]=env.fromElements(DeviceEvent(1,1,1),DeviceEvent(2,2,2))

我们通过按照第一个元素keyBy，然后按照第二个元素max

streamData.keyBy(0).max(1)

我们会得到结果：

1,1,1
1,2,1

如果我们同样按照第一个元素keyBy，但是按照第二个元素maxBy

streamData.keyBy(0).maxBy(1)

将会得到如下结果：

1,1,1
2,2,2

min和minBy同理。

调用方式：

flink内置聚合方法在scala里的调用方式有两种：

1. 传入position参数

def sum(position: Int): DataStream[T]

2. 传入field参数

def sum(field: String): DataStream[T]

代码演示：

演示这几种聚合方法

    // create DataStream from collection
    val data: DataStream[(String, Int, String)] = env.fromCollection(List(("device-1", 3, "1-3"), ("device-1", 1, "1-1"), ("device-2", 2, "2-2"), ("device-1", 2, "1-2")))
    val keyedStream = data.keyBy(0)
    keyedStream.sum(1).print("sum")
    keyedStream.min(1).print("min")
    keyedStream.max(1).print("max")
    keyedStream.minBy(1).print("minBy")
    keyedStream.maxBy(1).print("maxBy")

输出：

sum:3> (device-2,2,2-2)
max:3> (device-2,2,2-2)
min:1> (device-1,3,1-3)
maxBy:3> (device-2,2,2-2)
maxBy:1> (device-1,3,1-3)
minBy:1> (device-1,3,1-3)
sum:1> (device-1,3,1-3)
max:1> (device-1,3,1-3)
minBy:3> (device-2,2,2-2)
min:3> (device-2,2,2-2)
max:1> (device-1,3,1-3)
minBy:1> (device-1,1,1-1)
maxBy:1> (device-1,3,1-3)
max:1> (device-1,3,1-3)
min:1> (device-1,1,1-3)
minBy:1> (device-1,1,1-1)
sum:1> (device-1,4,1-3)
min:1> (device-1,1,1-3)
maxBy:1> (device-1,3,1-3)
sum:1> (device-1,6,1-3)

1.8 window

类型转换：

KeyedStream->WindowedStream

描述：

在KeyedStream上的窗口装换，Flink提供了四种窗口类型：Tumbling Windows(滚动窗口)、Sliding Windows(滑动窗口)、Session Windows(会话窗口)、Global Windows(全局窗口)，滚动窗口和滑动窗口分为基于时间的窗口和基于个数的窗口，而基于时间的窗口又分为事件时间窗口和处理时间窗口。关于窗口的更多内容会在下一篇文章中记录，这里只做简单演示。

调用方式：

window在scala里的调用方式有五种：

1. timeWindow(size: Time)，传入类型为Time的窗口大小，Flink会根据指定的time characteristic来判断是处理时间滚动窗口还是事件时间滚动窗口。

def timeWindow(size: Time): WindowedStream[T, K, TimeWindow]

2. timeWindow(size: Time, slide: Time)，传入类型为Time的窗口大小，以及类型为Time的窗口滑动大小，Flink同样会根据指定的time characteristic来判断是处理时间滑动窗口还是事件时间滑动窗口。

def timeWindow(size: Time, slide: Time): WindowedStream[T, K, TimeWindow]

3. countWindow(size: Long)，传入Long类型的窗口大小，Flink会应用为基于个数的滚动窗口。

def countWindow(size: Long): WindowedStream[T, K, GlobalWindow]

4. countWindow(size: Long, slide: Long)，传入Long类型的窗口大小，以及Long类型的窗口滑动带下，生成基于个数的滑动窗口。

def countWindow(size: Long, slide: Long): WindowedStream[T, K, GlobalWindow]

5. window，需要传入WindowAssigner实例

def window[W <: Window](assigner: WindowAssigner[_ >: T, W]): WindowedStream[T, K, W]

代码演示：

    // create data from collection
    val data = env.fromCollection(1 to 10)
    data.countWindowAll(5)

1.9 windowAll

类型转换：

DataStream->AllWindowedStream

描述：

在DataStream上的窗口装换，与window类似，只是window应用在KeyedStream上，需要先进行KeyBy操作，windowAll可以直接应用到DataStream上。

调用方式：

window在scala里的调用方式有五种：

1. timeWindowAll(size: Time)传入类型为Time的窗口大小，Flink会根据指定的time characteristic来判断是处理时间滚动窗口还是事件时间滚动窗口。

def timeWindowAll(size: Time): AllWindowedStream[T, TimeWindow] 

2. timeWindow(size: Time, slide: Time)，传入类型为Time的窗口大小，以及类型为Time的窗口滑动大小，Flink同样会根据指定的time characteristic来判断是处理时间滑动窗口还是事件时间滑动窗口。

def timeWindowAll(size: Time, slide: Time): AllWindowedStream[T, TimeWindow]

3. countWindow(size: Long)，传入Long类型的窗口大小，Flink会应用为基于个数的滚动窗口。

def countWindowAll(size: Long, slide: Long): AllWindowedStream[T, GlobalWindow]

4. countWindow(size: Long, slide: Long)，传入Long类型的窗口大小，以及Long类型的窗口滑动带下，生成基于个数的滑动窗口。

def countWindowAll(size: Long): AllWindowedStream[T, GlobalWindow]

5. windowAll，需要传入WindowAssigner实例

windowAll[W <: Window](assigner: WindowAssigner[_ >: T, W]): AllWindowedStream[T, W]

代码演示：

    // create data from collection
    val data = env.fromCollection((1 to 10).map(x => (x % 2, x)))
    // get KeyedDataStream
    val keyedStream = data.keyBy(0)
    val windowStream = keyedStream.countWindow(5)

输出：

windowStream.sum(1).print("countWindow")

countWindow:6> (1,25)
countWindow:6> (0,30)

1.10 WindowedStream apply

类型转换：

WindowedStream->DataStream

描述：

在WindowedStream上应用WindowFunction，当窗口被触发时调用指定的WindowFunction。

调用方式：

apply在scala里的调用方式有两种(其它预聚合的调用方式已废弃)：

1. 继承WindowFunction实现apply方法，并传入该实例

def apply[R: TypeInformation](
      function: WindowFunction[T, R, K, W]): DataStream[R]

2. 传入匿名函数

def apply[R: TypeInformation](
      function: (K, W, Iterable[T], Collector[R]) => Unit): DataStream[R]

代码演示：

在上面生成的windowStream调用apply

    val dataStream = windowedStream.apply((key, window, input, out: Collector[(Int, String, Int)]) => {
      var sum = 0
      input.foreach(element => {
        sum += element._2
      })
      out.collect(key.getField(0), window.toString, sum)
    })
    dataStream.print("simple apply")

输出：

simple apply:6> (1,GlobalWindow,25)
simple apply:6> (0,GlobalWindow,30)

1.11 WindowedStream reduce

类型转换：

WindowedStream->DataStream

描述：

在WindowedStream上应用ReduceFunction，与DataStream上的ReduceFunction相同，当窗口被触发时调用指定的ReduceFunction。

调用方式：

reduce在scala里的调用方式有六种，其中包括几种预聚合的方式调用：

1. 继承ReduceFunction实现reduce方法，并传入该实例

def reduce(function: ReduceFunction[T]): DataStream[T]

2. 传入匿名函数

def reduce(function: (T, T) => T): DataStream[T]

3. 传入ReduceFunction[T]作为预聚合器,当数据到达时调用，传入WindowFunction作为窗口函数，当触发窗口计算的时候调用。

  def reduce[R: TypeInformation](
  preAggregator: ReduceFunction[T],
  function: WindowFunction[T, R, K, W]): DataStream[R]

4. 传入匿名函数作为预聚合器,当数据到达时调用，传入WindowFunction作为窗口函数，当触发窗口计算的时候调用。

  def reduce[R: TypeInformation](
      preAggregator: (T, T) => T,
      windowFunction: (K, W, Iterable[T], Collector[R]) => Unit): DataStream[R]

5. 传入匿名函数作为预聚合器,当数据到达时调用，传入ProcessWindowFunction作为窗口函数，当触发窗口计算的时候调用。

  def reduce[R: TypeInformation](
      preAggregator: (T, T) => T,
      function: ProcessWindowFunction[T, R, K, W]): DataStream[R]

6. 传入ReduceFunction[T]作为预聚合器,当数据到达时调用，传入ProcessWindowFunction作为窗口函数，当触发窗口计算的时候调用。

  def reduce[R: TypeInformation](
      preAggregator: ReduceFunction[T],
      function: ProcessWindowFunction[T, R, K, W]): DataStream[R]

代码演示：

在上面生成的windowStream调用reduce

    windowedStream.reduce(new ReduceSum).print("function reduce")
    windowedStream.reduce((value1, value2) => (value2._1, value1._2 + value2._2)).print("simple reduce")

输出：

simple reduce:6> (1,25)
function reduce:6> (1,25)
simple reduce:6> (0,30)
function reduce:6> (0,30)

1.11 WindowedStream fold(deprecated)

deprecated

1.12 WindowedStream aggregations

类型转换：

WindowedStream->DataStream

描述：

WindowedStream上的聚合函数，与DataStream上的相同，也有sum、min、minBy、max、maxBy，参考DataStream上的aggregations。

调用方式：

参考DataStream上的aggregations。

代码演示：

    windowedStream.sum(0).print("sum")
    windowedStream.min(0).print("min")
    windowedStream.max(0).print("max")
    windowedStream.minBy(0).print("minBy")
    windowedStream.maxBy(0).print("maxBy")

输出：

maxBy:6> (1,1)
minBy:6> (1,1)
max:6> (1,1)
sum:6> (5,1)
min:6> (1,1)
maxBy:6> (0,2)
minBy:6> (0,2)
max:6> (0,2)
sum:6> (0,2)
min:6> (0,2)

1.13 union

类型转换：

DataStream[T],DataStream[T],*->DataStream[T]

描述：

将两个或更多的流进行合并，要求数据类的类型是一直的。

调用方式：

该算子调用方式只有一种：

1. 传入不定长的DataStream[T]

def union(dataStreams: DataStream[T]*): DataStream[T]

代码演示：

先通过1 到10的集合生成一个DataStream，并通过filter操作，过滤出奇数

    val stream1 = env.fromCollection(1 to 10)
    val stream2 = env.fromCollection(101 to 110)
    stream1.union(stream2).print("union")

输出：

union:4> 1
union:7> 101
union:4> 9
union:2> 104
union:3> 105
union:5> 107
union:5> 2
union:1> 103
union:6> 108
union:1> 6
union:6> 3
union:8> 102
union:5> 10
union:2> 7
union:4> 106
union:3> 8
union:7> 109
union:8> 110
union:7> 4
union:8> 5

1.13 join

类型转换：

DataStream[T],DataStream[T2]->JoinedStreams[T, T2]

描述：

将两个按照指定的条件流进行join，两个流的数据类型可以不同，需要与窗口函数结合使用。

调用方式：

该算子调用方式只有一种：

1. 传入需要Join的DataStream[T2]

def join[T2](otherStream: DataStream[T2]): JoinedStreams[T, T2] 

代码演示：

定义两个流，一个是从集合中创建，另一个从socket中创建。

    val stream1 = env.fromCollection(List(("device-1", 1), ("device-2", 2)))
    val stream2 = env.socketTextStream("localhost", 9090).map(x => {
      val token = x.split(" ")
      (token(0), token(1))
    })
    val joinedStream: JoinedStreams[(String, Int), (String, String)]#Where[String]#EqualTo = stream1.join(stream2).where(_._1).equalTo(_._1)

    joinedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
      .apply((first, second) => (first._1, second._2, first._2)).print("join")

使用nc -lk 9090监听9090端口，启动程序后，发送数据：

输出：

union:4> 1
union:7> 101
union:4> 9
union:2> 104
union:3> 105
union:5> 107
union:5> 2
union:1> 103
union:6> 108
union:1> 6
union:6> 3
union:8> 102
union:5> 10
union:2> 7
union:4> 106
union:3> 8
union:7> 109
union:8> 110
union:7> 4
union:8> 5