Flink学习20：聚合算子(sum,max,min)

常见的聚合算子 sum,max,min等

聚合算子可以在在keyedStream 流上进行滚动的聚合&＃xff08;即累计的操作&＃xff09;&＃xff0c;而且同一个 keyedStream 流上只能调用一次 聚合算子

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentobject aggregationTest {//defined the dataSource&＃39;s typecase class StockPrice(stockId:String, timeStamp:Long, price:Double)def main(args: Array[String]): Unit &＃61; {//create envval env &＃61; StreamExecutionEnvironment.getExecutionEnvironment//generate dsval stockList &＃61; List(StockPrice("stock_1", 66666, 1), StockPrice("stock_1", 8888, 2), StockPrice("stock_2", 77777, 1), StockPrice("stock_2", 999, 3), StockPrice("stock_3", 3333, 1))val ds &＃61; env.fromCollection(stockList)//transformationval keyedStream &＃61; ds.keyBy("stockId")val sumedStream &＃61; keyedStream.sum(2)sumedStream.print()env.execute()}}

输出结果&＃xff1a;

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment//defined the dataSource&＃39;s type
case class StockPrice(stockId:String, timeStamp:Long, price:Double)object maxTest {def main(args: Array[String]): Unit &＃61; {//create envval env &＃61; StreamExecutionEnvironment.getExecutionEnvironment//generate dsval stockList &＃61; List(StockPrice("stock_1", 66666, 1), StockPrice("stock_1", 8888, 2), StockPrice("stock_2", 77777, 1), StockPrice("stock_2", 999, 3), StockPrice("stock_3", 3333, 1))val ds &＃61; env.fromCollection(stockList)//transformationval keyedStream &＃61; ds.keyBy("stockId")val maxedStream &＃61; keyedStream.max(2)maxedStream.print()env.execute()}}

输出结果&＃xff1a;