转载Flink Keyed Windows函数：ProcessWindowFunction ，ReduceFunction ，AggregateFunction ，FoldFunction

墨蓝 2024-04-17 17:48 63阅读 0赞

此文案例问转载：[https://www.cnblogs.com/maoxiangyi/p/10978028.html][https_www.cnblogs.com_maoxiangyi_p_10978028.html]

1. ProcessWindowFunction  效率低，整个窗口的数据都会被传递给算子，代码如下：

>     import org.apache.flink.api.common.functions.AggregateFunction
>     import org.apache.flink.api.scala._
>     import org.apache.flink.streaming.api.functions.source.SourceFunction
>     import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
>     import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
>     import org.apache.flink.streaming.api.windowing.time.Time
>     import org.apache.flink.streaming.api.windowing.windows.TimeWindow
>     import org.apache.flink.util.Collector
>     
>     /**
>      * @author: create by maoxiangyi
>      * @version: v1.0
>      * @description: window
>      * @date:2019 /6/4
>      */
>     object ProcessWordCount {
>      def main(args: Array[String]): Unit = {
>       //设置环境
>       val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment()
>       //设置数据源
>       env.addSource(new SourceFunction[String] {
>        override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
>         while (true) {
>          ctx.collect("hello hadoop hello storm hello spark")
>          Thread.sleep(1000)
>         }
>        }
>     
>        override def cancel(): Unit = {}
>       })
>        //计算逻辑
>        .flatMap(_.split(" "))
>        .map((_, 1))
>        .keyBy(_._1)
>        .timeWindow(Time.seconds(10), Time.seconds(10))
> 
>     .process(new ProcessWindowFunction[(String, Int), (String, Int), String, TimeWindow] {
>         override def process(key: String, context: Context, elements: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {
>          var value = 0;
>          elements.foreach(kv => {
>           value = value + kv._2
>          })
>          out.collect(key, value)
>         }
>        })
>        .print().setParallelism(1)
>       env.execute("word count")
>      }
>     }

2，ReduceFunction  简单易用，推荐使用，代码：

>     package com.coder.flink.core.aaa_1
>     
>     import org.apache.flink.streaming.api.functions.source.SourceFunction
>     import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
>     import org.apache.flink.streaming.api.windowing.time.Time
>     import org.apache.flink.api.scala._
>     
>     
>     object ReduceWordCount {
>       def main(args: Array[String]): Unit = {
>         //设置环境
>         val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment()
>         //设置数据源
>         env.addSource(new SourceFunction[String] {
>           override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
>             while (true) {
>               ctx.collect("hello hadoop hello storm hello spark")
>               Thread.sleep(1000)
>             }
>           }
>     
>           override def cancel(): Unit = {}
>         })
>           //计算逻辑
>           .flatMap(_.split(" "))
>           .map((_, 1))
>           .keyBy(_._1)
>           .timeWindow(Time.seconds(10), Time.seconds(10))
>           .reduce((x, y) => {
>             (x._1, x._2 + y._2)
>           })
>           //输出结果
>           .addSink(x => {
>           print(x + "\n")
>         })
>         //提交任务
>         env.execute("word count")
>       }
>     }

3. AggregateFunction  复杂逻辑 可以优化使用：

>     import org.apache.flink.api.common.functions.AggregateFunction
>     import org.apache.flink.streaming.api.functions.source.SourceFunction
>     import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
>     import org.apache.flink.streaming.api.windowing.time.Time
>     import org.apache.flink.api.scala._
>     
>     /**
>      * @author: create by maoxiangyi
>      * @version: v1.0
>      * @description: window
>      * @date:2019 /6/4
>      */
>     object AggregateWordCount {
>      def main(args: Array[String]): Unit = {
>       //设置环境
>       val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment()
>       //设置数据源
>       env.addSource(new SourceFunction[String] {
>        override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
>         while (true) {
>          ctx.collect("hello hadoop hello storm hello spark")
>          Thread.sleep(1000)
>         }
>        }
>     
>        override def cancel(): Unit = {}
>       })
>        //计算逻辑
>        .flatMap(_.split(" "))
>        .map((_, 1))
>        .keyBy(_._1)
>        .timeWindow(Time.seconds(10), Time.seconds(10))
> 
>      
> 
>     .aggregate(new AggregateFunction[(String, Int), (String, Int), (String, Int)] {
>         override def createAccumulator(): (String, Int) = {
>          ("", 0)
>         }
>         override def add(value: (String, Int), accumulator: (String, Int)): (String, Int) = {
>          (value._1, accumulator._2 + value._2)
>         }
>         override def getResult(accumulator: (String, Int)): (String, Int) = accumulator
>     
>         override def merge(a: (String, Int), b: (String, Int)): (String, Int) = {
>          (a._1, a._2 + b._2)
>         }
>        }).print().setParallelism(1)
>       env.execute("word count")
>      }
>     }

4. FoldFunction  过时API，不够友好，代码：

>     import org.apache.flink.api.common.functions.{AggregateFunction, FoldFunction}
>     import org.apache.flink.api.scala._
>     import org.apache.flink.streaming.api.functions.source.SourceFunction
>     import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, WindowedStream}
>     import org.apache.flink.streaming.api.windowing.time.Time
>     import org.apache.flink.streaming.api.windowing.windows.TimeWindow
>     
>     /**
>      * @author: create by maoxiangyi
>      * @version: v1.0
>      * @description: window
>      * @date:2019 /6/4
>      */
>     object FoldWordCount {
>      def main(args: Array[String]): Unit = {
>       //设置环境
>       val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment()
>       //设置数据源
>       val window: WindowedStream[(String, Int), String, TimeWindow] = env.addSource(new SourceFunction[String] {
>        override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
>         while (true) {
>          ctx.collect("hello hadoop hello storm hello spark")
>          Thread.sleep(1000)
>         }
>        }
>     
>        override def cancel(): Unit = {}
>       })
>        //计算逻辑
>        .flatMap(_.split(" "))
>        .map((_, 1))
>        .keyBy(_._1)
>        .timeWindow(Time.seconds(10), Time.seconds(10))
> 
>      
> 
>     window.fold(("", 0), new FoldFunction[(String, Int), (String, Int)] {
>        override def fold(accumulator: (String, Int), value: (String, Int)): (String, Int) = {
>         (value._1, value._2 + accumulator._2)
>        }
>       }).print()
>     
>       env.execute("word count")
>      }
>     }

5，,no-key类型  ProcessAllWindowFunction  将所有的数据放到一个窗口，性能低下，不常用 ：

>     import org.apache.flink.api.common.functions.FlatMapFunction
>     import org.apache.flink.streaming.api.functions.source.SourceFunction
>     import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
>     import org.apache.flink.streaming.api.scala.function.ProcessAllWindowFunction
>     import org.apache.flink.streaming.api.windowing.time.Time
>     import org.apache.flink.streaming.api.windowing.windows.TimeWindow
>     import org.apache.flink.util.Collector
>     import org.apache.flink.api.scala._
>     
>     import scala.collection.mutable
>     
>      
>     object ProcessAllWindowWordCount {
>      def main(args: Array[String]): Unit = {
>       //设置环境
>       val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment()
>       //设置数据源
>       env.addSource(new SourceFunction[String] {
>        override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
>         while (true) {
>          ctx.collect("hello hadoop hello storm hello spark")
>          Thread.sleep(1000)
>         }
>        }
>     
>        override def cancel(): Unit = {}
>       })
>        //计算逻辑
>        .flatMap(_.split(" "))
>        .map((_, 1))
>        .timeWindowAll(Time.seconds(10), Time.seconds(10))
>        
>        .process(new ProcessAllWindowFunction[(String, Int), mutable.Map[String, Int], TimeWindow] {
>         override def process(context: Context, elements: Iterable[(String, Int)], out: Collector[mutable.Map[String, Int]]): Unit = {
>          val wordCountMap = mutable.Map[String, Int]()
>          elements.foreach(kv => {
>           wordCountMap.put(kv._1, wordCountMap.get(kv._1).getOrElse(0) + kv._2)
>          })
>          out.collect(wordCountMap)
>         }
>        }).flatMap(new FlatMapFunction[mutable.Map[String, Int], (String, Int)] {
>        override def flatMap(value: mutable.Map[String, Int], out: Collector[(String, Int)]): Unit = {
>         value.foreach(out.collect(_))
>        }
>       })
>        .print()
>       //提交任务
>       env.execute("word count")
>      }
>     }

[https_www.cnblogs.com_maoxiangyi_p_10978028.html]: https://www.cnblogs.com/maoxiangyi/p/10978028.html