Flink流计算之聚合函数

雨点打透心脏的1/2处 2022-11-25 01:59 272阅读 0赞

## 一.Reduce功能 ##

ReduceFunction指定如何将输入中的两个元素组合在一起以产生相同类型的输出元素。Flink使用ReduceFunction来逐步聚合窗口的元素。

ReduceFunction可以定义像这样使用：

val input: DataStream[(String, Long)] = ...
    
    input
        .keyBy(<key selector>)
        .window(<window assigner>)
        .reduce {  (v1, v2) => (v1._1, v1._2 + v2._2) }

## 二.聚合函数 ##

AggregateFunction是一般化版本的ReduceFunction，其具有三种类型：输入类型【IN】，累加器【ACC】和一个输出类型【OUT】。输入类型是输入流中元素的类型，并且AggregateFunction具有将一个输入元素添加到累加器的方法。该接口还具有创建初始累加器，将两个累加器合并为一个累加器以及从累加器提取输出的方法。

与ReduceFunction一样，Flink将在窗口输入元素到达时增量地对其进行聚合。

AggregateFunction可以被定义并这样使用：

/** * The accumulator is used to keep a running sum and a count. The [getResult] method * computes the average. */
    class AverageAggregate extends AggregateFunction[(String, Long), (Long, Long), Double] { 
      override def createAccumulator() = (0L, 0L)
    
      override def add(value: (String, Long), accumulator: (Long, Long)) =
        (accumulator._1 + value._2, accumulator._2 + 1L)
    
      override def getResult(accumulator: (Long, Long)) = accumulator._1 / accumulator._2
    
      override def merge(a: (Long, Long), b: (Long, Long)) =
        (a._1 + b._1, a._2 + b._2)
    }
    
    val input: DataStream[(String, Long)] = ...
    
    input
        .keyBy(<key selector>)
        .window(<window assigner>)
        .aggregate(new AverageAggregate)

上面的示例计算窗口中元素的第二个字段的平均值。

## 三.FoldFunction ##

FoldFunction指定如何将窗口的输入元素与输出类型的元素组合。FoldFunction递增地被添加到窗口并且输出每个元素。第一个元素与输出类型的预定义初始值组合。

FoldFunction可以定义像这样使用：

val input: DataStream[(String, Long)] = ...
    
    input
        .keyBy(<key selector>)
        .window(<window assigner>)
        .fold("") {  (acc, v) => acc + v._2 }

注意 fold()不能与会话窗口或其他可合并窗口一起使用。

## 四.ProcessWindowFunction ##

ProcessWindowFunction获得一个Iterable，该Iterable包含窗口的所有元素，以及一个Context对象，该对象可以访问时间和状态信息，从而使其比其他窗口函数更具灵活性。这是以性能和资源消耗为代价的，因为不能增量聚合元素，而是需要在内部对其进行缓冲，直到将窗口视为已准备好进行处理为止。

ProcessWindowFunction的代码如下：

abstract class ProcessWindowFunction[IN, OUT, KEY, W <: Window] extends Function { 
    
      /** * Evaluates the window and outputs none or several elements. * * @param key The key for which this window is evaluated. * @param context The context in which the window is being evaluated. * @param elements The elements in the window being evaluated. * @param out A collector for emitting elements. * @throws Exception The function may throw exceptions to fail the program and trigger recovery. */
      def process(
          key: KEY,
          context: Context,
          elements: Iterable[IN],
          out: Collector[OUT])
    
      /** * The context holding window metadata */
      abstract class Context { 
        /** * Returns the window that is being evaluated. */
        def window: W
    
        /** * Returns the current processing time. */
        def currentProcessingTime: Long
    
        /** * Returns the current event-time watermark. */
        def currentWatermark: Long
    
        /** * State accessor for per-key and per-window state. */
        def windowState: KeyedStateStore
    
        /** * State accessor for per-key global state. */
        def globalState: KeyedStateStore
      }
    
    }

注意：key参数是通过KeySelector提取，被指定的keyBy()调用。如果是元组索引键或字符串字段引用，则始终使用此键类型，Tuple类型必须手动将其强制转换为正确大小的元组以提取键字段。

ProcessWindowFunction可以定义这样使用：

val input: DataStream[(String, Long)] = ...
    
    input
      .keyBy(_._1)
      .timeWindow(Time.minutes(5))
      .process(new MyProcessWindowFunction())
    
    /* ... */
    
    class MyProcessWindowFunction extends ProcessWindowFunction[(String, Long), String, String, TimeWindow] { 
    
      def process(key: String, context: Context, input: Iterable[(String, Long)], out: Collector[String]): () = { 
        var count = 0L
        for (in <- input) { 
          count = count + 1
        }
        out.collect(s"Window ${context.window} count: $count")
      }
    }

显示了一个ProcessWindowFunction计算窗口中元素的方法。另外，窗口功能将有关窗口的信息添加到输出中。

注意：ProcessWindowFunction用于简单的聚合【例如count】效率很低！

## 五.具有增量聚合的ProcessWindowFunction ##

ProcessWindowFunction可以将ReduceFunction、AggregateFunction或FoldFunction元素到达窗口时对其进行递增聚合。当窗口关闭时，ProcessWindowFunction将提供汇总结果。这样一来，ProcessWindowFunction就可以增量计算窗口，同时可以访问其他窗口的元信息。

注意：也可以使用旧版WindowFunction而不是 ProcessWindowFunction用于增量窗口聚合。

**具有ReduceFunction的增量窗口聚合**  
以下示例显示了如何将增量ReduceFunction与组合ProcessWindowFunction以返回窗口中最小的事件以及窗口的开始时间。

val input: DataStream[SensorReading] = ...
    
    input
      .keyBy(<key selector>)
      .timeWindow(<duration>)
      .reduce(
        (r1: SensorReading, r2: SensorReading) => {  if (r1.value > r2.value) r2 else r1 },
        ( key: String,
          context: ProcessWindowFunction[_, _, _, TimeWindow]#Context,
          minReadings: Iterable[SensorReading],
          out: Collector[(Long, SensorReading)] ) =>
          { 
            val min = minReadings.iterator.next()
            out.collect((context.window.getStart, min))
          }
      )

**具有AggregateFunction的增量窗口聚合**  
以下示例显示了如何将增量AggregateFunction与ProcessWindowFunction组合以计算平均值，并与平均值一起输出键和窗口。

val input: DataStream[(String, Long)] = ...
    
    input
      .keyBy(<key selector>)
      .timeWindow(<duration>)
      .aggregate(new AverageAggregate(), new MyProcessWindowFunction())
    
    // Function definitions
    
    /** * The accumulator is used to keep a running sum and a count. The [getResult] method * computes the average. */
    class AverageAggregate extends AggregateFunction[(String, Long), (Long, Long), Double] { 
      override def createAccumulator() = (0L, 0L)
    
      override def add(value: (String, Long), accumulator: (Long, Long)) =
        (accumulator._1 + value._2, accumulator._2 + 1L)
    
      override def getResult(accumulator: (Long, Long)) = accumulator._1 / accumulator._2
    
      override def merge(a: (Long, Long), b: (Long, Long)) =
        (a._1 + b._1, a._2 + b._2)
    }
    
    class MyProcessWindowFunction extends ProcessWindowFunction[Double, (String, Double), String, TimeWindow] { 
    
      def process(key: String, context: Context, averages: Iterable[Double], out: Collector[(String, Double)]): () = { 
        val average = averages.iterator.next()
        out.collect((key, average))
      }
    }

**具有FoldFunction的增量窗口聚合**  
以下示例显示了如何将增量FoldFunction与ProcessWindowFunction组合以提取窗口中的事件数，并还返回窗口的键和结束时间。

val input: DataStream[SensorReading] = ...
    
    input
     .keyBy(<key selector>)
     .timeWindow(<duration>)
     .fold (
        ("", 0L, 0),
        (acc: (String, Long, Int), r: SensorReading) => {  ("", 0L, acc._3 + 1) },
        ( key: String,
          window: TimeWindow,
          counts: Iterable[(String, Long, Int)],
          out: Collector[(String, Long, Int)] ) =>
          { 
            val count = counts.iterator.next()
            out.collect((key, window.getEnd, count._3))
          }
      )