Flink学习20：聚合算子(sum,max,min)-Toy模板网

这篇具有很好参考价值的文章主要介绍了Flink学习20：聚合算子(sum,max,min)。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

1.聚合算子简介

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

聚合算子可以在在keyedStream 流上进行滚动的聚合（即累计的操作），而且同一个 keyedStream 流上只能调用一次聚合算子

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

文章来源地址https://www.toymoban.com/news/detail-471425.html

sum 示例：

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object aggregationTest {

//defined the dataSource's type
  case class StockPrice(stockId:String, timeStamp:Long, price:Double)


  def main(args: Array[String]): Unit = {

    //create env
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //generate ds

    val stockList = 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 = env.fromCollection(stockList)

    //transformation

    val keyedStream = ds.keyBy("stockId")

    val sumedStream = keyedStream.sum(2)

    sumedStream.print()

    env.execute()

  }

}

输出结果：

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

max示例：

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

//defined the dataSource's type
case class StockPrice(stockId:String, timeStamp:Long, price:Double)

object maxTest {

  def main(args: Array[String]): Unit = {

    //create env
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //generate ds
    val stockList = 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 = env.fromCollection(stockList)

    //transformation
    val keyedStream = ds.keyBy("stockId")

    val maxedStream = keyedStream.max(2)

    maxedStream.print()

    env.execute()

  }


}