update

Author: confucianzuoyuan

Flink0105/src/main/scala/com/atguigu/day8/AggregateFunctionExample.scala

@@ -0,0 +1,79 @@
+package com.atguigu.day8
+
+import com.atguigu.day2.SensorSource
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.EnvironmentSettings
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.types.Row
+
+object AggregateFunctionExample {
+  def main(args: Array[String]): Unit = {
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    env.setParallelism(1)
+
+    val stream = env.addSource(new SensorSource).filter(_.id.equals("sensor_1"))
+
+    val settings = EnvironmentSettings
+      .newInstance()
+      .useBlinkPlanner()
+      .inStreamingMode()
+      .build()
+
+    val tEnv = StreamTableEnvironment.create(env, settings)
+
+    val avgTemp = new AvgTemp()
+
+    // table api
+    val table = tEnv.fromDataStream(stream, 'id, 'timestamp as 'ts, 'temperature as 'temp)
+
+    table
+        .groupBy('id)
+        .aggregate(avgTemp('temp) as 'avgTemp)
+        .select('id, 'avgTemp)
+        .toRetractStream[Row]
+//        .print()
+
+    // sql 写法
+    tEnv.createTemporaryView("sensor", table)
+
+    tEnv.registerFunction("avgTemp", avgTemp)
+
+    tEnv.sqlQuery(
+      """
+        |SELECT
+        | id, avgTemp(temp)
+        | FROM
+        | sensor
+        | GROUP BY id""".stripMargin
+    )
+        .toRetractStream[Row].print()
+
+
+    env.execute()
+  }
+
+  // 累加器的类型
+  class AvgTempAcc {
+    var sum: Double = 0.0
+    var count: Int = 0
+  }
+
+  // 第一个泛型是温度值的类型
+  // 第二个泛型是累加器的类型
+  class AvgTemp extends AggregateFunction[Double, AvgTempAcc] {
+    // 创建累加器
+    override def createAccumulator(): AvgTempAcc = new AvgTempAcc
+
+    // 累加规则
+    def accumulate(acc: AvgTempAcc, temp: Double): Unit = {
+      acc.sum += temp
+      acc.count += 1
+    }
+
+    override def getValue(accumulator: AvgTempAcc): Double = {
+      accumulator.sum / accumulator.count
+    }
+  }
+}

Flink0105/src/main/scala/com/atguigu/day8/TableAggregateFunctionExample.scala

@@ -0,0 +1,83 @@
+package com.atguigu.day8
+
+import com.atguigu.day2.SensorSource
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.EnvironmentSettings
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.functions.{AggregateFunction, TableAggregateFunction}
+import org.apache.flink.types.Row
+import org.apache.flink.util.Collector
+
+object TableAggregateFunctionExample {
+  def main(args: Array[String]): Unit = {
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    env.setParallelism(1)
+
+    val stream = env.addSource(new SensorSource).filter(_.id.equals("sensor_1"))
+
+    val settings = EnvironmentSettings
+      .newInstance()
+      .useBlinkPlanner()
+      .inStreamingMode()
+      .build()
+
+    val tEnv = StreamTableEnvironment.create(env, settings)
+
+    val top2Temp = new Top2Temp()
+
+    // table api
+    val table = tEnv.fromDataStream(stream, 'id, 'timestamp as 'ts, 'temperature)
+
+    table
+        .groupBy('id)
+        .flatAggregate(top2Temp('temperature) as ('temp, 'rank))
+        .select('id, 'temp, 'rank)
+        .toRetractStream[Row]
+        .print()
+
+    // sql
+    tEnv.createTemporaryView("t", table)
+
+    tEnv.registerFunction("top2Temp", top2Temp)
+
+//    tEnv
+//        .sqlQuery(
+//          """
+//            |SELECT id, top2Temp(temperature)
+//            | FROM t GROUP BY id""".stripMargin)
+//        .toRetractStream[Row]
+//        .print()
+
+
+    env.execute()
+  }
+
+  // 累加器
+  class Top2TempAcc {
+    var highestTemp: Double = Double.MinValue
+    var secondHighestTemp: Double = Double.MinValue
+  }
+
+  // 第一个泛型是输出：（温度值，排名）
+  // 第二个泛型是累加器
+  class Top2Temp extends TableAggregateFunction[(Double, Int), Top2TempAcc] {
+    override def createAccumulator(): Top2TempAcc = new Top2TempAcc
+
+    // 函数名必须是accumulate
+    def accumulate(acc: Top2TempAcc, temp: Double): Unit = {
+      if (temp > acc.highestTemp) {
+        acc.secondHighestTemp = acc.highestTemp
+        acc.highestTemp = temp
+      } else if (temp > acc.secondHighestTemp) {
+        acc.secondHighestTemp = temp
+      }
+    }
+
+    // 函数名必须是emitValue，用来发射计算结果
+    def emitValue(acc: Top2TempAcc, out: Collector[(Double, Int)]): Unit = {
+      out.collect(acc.highestTemp, 1)
+      out.collect(acc.secondHighestTemp, 2)
+    }
+  }
+}