Flink自定义aggregate聚合函数的步骤

第一步：将dataStream转换城windowedStream

        // 从kafka读取数据
        val inputStream = env.addSource(new FlinkKafkaConsumer[String]("hotitems", new SimpleStringSchema(), properties))
            .map(data => {
                val dataArray = data.split(",")
                UserBehavior(dataArray(0).toLong, dataArray(1).toLong, dataArray(2).toInt, dataArray(3), dataArray(4).toLong)
            })
            .assignAscendingTimestamps(_.timestamp * 1000L)
        
        // 对数据进行窗口聚合处理
        val aggStream: DataStream[ItemViewCount] = inputStream
            .filter(_.behavior == "pv") // 过滤出pv数据
            .keyBy(_.itemId)
            .timeWindow(Time.hours(1), Time.minutes(5)) // 开窗进行统计
            .aggregate(new CountAgg(), new WindowCountResult()) // 聚合出当前商品在时间窗口内的统计数量

第二步：自定义聚合函数

// 自定义的预聚合函数，来一条数据就加一
class CountAgg() extends AggregateFunction[UserBehavior, Long, Long] {
    //add方法为累加器累加的方法，这里为最简单的+1操作
    override def add(value: UserBehavior, accumulator: Long): Long = accumulator + 1
    //初始化累加值
    override def createAccumulator(): Long = 0L
    //最后返回那个值，这里为accumulator
    override def getResult(accumulator: Long): Long = accumulator
    //分区处理的归并操作，这里将所有并处理的结果相加
    override def merge(a: Long, b: Long): Long = a + b
}

第三部：自定义窗口函数

// 自定义window function
class WindowCountResult() extends WindowFunction[Long, ItemViewCount, Long, TimeWindow] {
    //Long类型的Key为上一步的自定义累加器的返回值
    //Window为差给你扣类型，第一步中的没窗口类型，TimeWindow
    //input为接收的数据类型，此处为Long类型的迭代器
    //out为此方法返回的类型，此处为ItemViewCount样例类对象的集合
    override def apply(key: Long, window: TimeWindow, input: Iterable[Long], out: Collector[ItemViewCount]): Unit = {
        //调用ItemViewCount样例类对象的构造器，依次构造出ItemViewCount样例类并返回
        out.collect(ItemViewCount(key, window.getEnd, input.iterator.next()))
    }
}