Flink实战:全局TopN分析与实现

野性酷女 2023-06-13 15:28 23阅读 0赞

点击蓝字

![640?wx\_fmt=png][640_wx_fmt_png]

关注~~

*  按照区域areaId+商品gdsId分组，计算每个分组的累计销售额
 *  将得到的区域areaId+商品gdsId维度的销售额按照区域areaId分组，然后求得TopN的销售额商品，并且定时更新输出

与窗口TopN不同，全局TopN没有时间窗口的概念，也就没有时间的概念，因此使用ProcessingTime语义即可，并且也不能再使用Window算子来操作，但是在这个过程中需要完成数据累加操作与定时输出功能，选择ProcessFunction函数来完成，使用State保存中间结果数据，保证数据一致性语义，使用定时器来完成定时输出功能。

### 销售额统计 ###

对数据流按照区域areaId+商品gdsId分组，不断累加销售额保存起来，然后输出到下游。

val env =StreamExecutionEnvironment.getExecutionEnvironment	
        env.setParallelism(1)	
        val kafkaConfig =newProperties();	
        kafkaConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");	
        kafkaConfig.put(ConsumerConfig.GROUP_ID_CONFIG,"test1");	
        val consumer =newFlinkKafkaConsumer011[String]("topic1",newSimpleStringSchema(), kafkaConfig)	
        val orderStream = env.addSource(consumer)	
                   .map(x =>{	
                          val a = x.split(",")	
                          Order(a(0), a(1).toLong, a(2), a(3).toDouble, a(4))	
                        })	
        val salesStream=orderStream.keyBy(x =>{	
                      x.areaId +"_"+ x.gdsId	
                 }).process(newKeyedProcessFunction[String,Order,GdsSales](){	
          var orderState:ValueState[Double]= _	
          var orderStateDesc:ValueStateDescriptor[Double]= _	
         override def open(parameters:Configuration):Unit={	
            orderStateDesc =newValueStateDescriptor[Double]("order-state",TypeInformation.of(classOf[Double]))	
            orderState = getRuntimeContext.getState(orderStateDesc)	
    }	
    override def processElement(value:Order, ctx:KeyedProcessFunction[String,Order,GdsSales]#Context,out:Collector[GdsSales]):Unit={	
            val currV = orderState.value()	
            if(currV ==null){	
                 orderState.update(value.amount)	
           }else{	
              val newV = currV + value.amount	
              orderState.update(newV)	
            }	
           out.collect(GdsSales.of(value.areaId, value.gdsId, orderState.value(), value.orderTime))	
    }	
    })

使用keyBy按照areaId+gdsId来分组，然后使用KeyedProcessFunction来完成累加操作。在KeyedProcessFunction里面定义了一个ValueState来保存每个分组的销售额，processElement完成销售额累加操作，并且不断更新ValueState与collect输出。说明：这里使用ValueState来完成累加过程显得比较繁琐，可以使用ReducingState来替代，这里只是为了表现出累加这个过程。

### 区域TopN计算 ###

1.  如果TreeSet\[GdsSales\]包含该商品的销售额数据，则需要更新该商品销售额，这个过程包含判断商品gdsId是否存在与移除该GdsSales对象功能，但是TreeSet不具备直接判断gdsId是否存在功能，那么可以使用一种额外的数据结构Map, key为商品gdsId, value为商品销售额数据GdsSales，该value对应TreeSet\[GdsSales\]中数据
2.  如果TreeSet\[GdsSales\]包含该商品的销售额数据，当TreeSet里面的数据到达N, 就获取第一个节点数据(最小值)与当前需要插入的数据进行比较，如果比其大，则直接舍弃，如果比其小，那么就将TreeSet中第一个节点数据删除，插入新的数据实现代码如下：

salesStream.keyBy(_.getAreaId)	
    .process(newKeyedProcessFunction[String,GdsSales,Void]{	
    var topState:ValueState[java.util.TreeSet[GdsSales]]= _	
    var topStateDesc:ValueStateDescriptor[java.util.TreeSet[GdsSales]]= _	
    var mappingState:MapState[String,GdsSales]= _	
    var mappingStateDesc:MapStateDescriptor[String,GdsSales]= _	
            val interval:Long=60000	
            val N:Int=3	
    override def open(parameters:Configuration):Unit={	
              topStateDesc =newValueStateDescriptor[java.util.TreeSet[GdsSales]]("top-state",TypeInformation.of(classOf[java.util.TreeSet[GdsSales]]))	
              topState = getRuntimeContext.getState(topStateDesc)	
              mappingStateDesc =newMapStateDescriptor[String,GdsSales]("mapping-state",TypeInformation.of(classOf[String]),TypeInformation.of(classOf[GdsSales]))	
              mappingState = getRuntimeContext.getMapState(mappingStateDesc)	
    }	
    override def processElement(value:GdsSales, ctx:KeyedProcessFunction[String,GdsSales,Void]#Context,out:Collector[Void]):Unit={	
              val top = topState.value()	
              if(top ==null){	
                val topMap: util.TreeSet[GdsSales]=new util.TreeSet[GdsSales](newComparator[GdsSales]{	
              override def compare(o1:GdsSales, o2:GdsSales):Int=(o1.getAmount - o2.getAmount).toInt	
    })	
                topMap.add(value)	
                topState.update(topMap)	
                mappingState.put(value.getGdsId, value)	
         }else{	
                mappingState.contains(value.getGdsId) match { 	
              case true=>{//已经存在该商品的销售数据	
                    val oldV = mappingState.get(value.getGdsId)	
                    mappingState.put(value.getGdsId, value)	
                    val values = topState.value()	
                    values.remove(oldV)	
                    values.add(value)//更新旧的商品销售数据	
                    topState.update(values)	
               }	
          case false=>{//不存在该商品销售数据	
                    if(top.size()>= N){//已经达到N 则判断更新	
                      val min = top.first()	
                        if(value.getAmount > min.getAmount){	
                        top.pollFirst()	
                        top.add(value)	
                        mappingState.put(value.getGdsId, value)	
                        topState.update(top)	
                           }	
                  }else{//还未到达N则直接插入	
                         top.add(value)	
                         mappingState.put(value.getGdsId, value)	
                         topState.update(top)	
    }	
    }}}}	
    })

在open中定义个两个state:ValueState与MapState, ValueState保存该区域下的TopN商品销售数据GdsSales,MapState保存了商品gdsId与商品销售数据GdsSale的对应关系。在processElement中，首先会判断ValueState是否为空，如果为空则定义按照销售额比较升序排序的Comparator 的TreeSet，则走更新逻辑判断。

### 定时输出 ###

var fireState:ValueState[Long]= _	
    var fireStateDesc:ValueStateDescriptor[Long]= _	
    //放在open方法中	
    fireStateDesc =newValueStateDescriptor[Long]("fire-time",TypeInformation.of(classOf[Long]))	
    fireState = getRuntimeContext.getState(fireStateDesc)

定义了一个ValueState，保存每一次触发的时间，不使用ReducingState是因为没有Window里面在使用SessionWindow的合并机制。

//放在processElement里面	
    val currTime = ctx.timerService().currentProcessingTime()	
    //1min输出一次	
    if(fireState.value()==null){	
                val start = currTime -(currTime % interval)	
                val nextFireTimestamp = start + interval	
                ctx.timerService().registerProcessingTimeTimer(nextFireTimestamp)	
                fireState.update(nextFireTimestamp)	
    }

对于每一个区域areaId(key)在processElement只需要注册一次即可。

override def onTimer(timestamp:Long, ctx:KeyedProcessFunction[String,GdsSales,Void]#OnTimerContext,out:Collector[Void]):Unit={	
              println(timestamp +"===")	
              topState.value().foreach(x =>{	
                println(x)	
              })	
              val fireTimestamp = fireState.value()	
              if(fireTimestamp !=null&&(fireTimestamp == timestamp)){	
                      fireState.clear()	
                      fireState.update(timestamp + interval)	
                     ctx.timerService().registerProcessingTimeTimer(timestamp + interval)	
             }	
    }

onTimer定时输出，并且注册下一个触发的时间点。

### 测试 ###

准备数据

//2019-11-16 21:25:10	
    orderId01,1573874530000,gdsId03,300,beijing	
    orderId02,1573874540000,gdsId01,100,beijing	
    orderId02,1573874540000,gdsId04,200,beijing	
    orderId02,1573874540000,gdsId02,500,beijing	
    orderId01,1573874530000,gdsId01,300,beijing

等到2019-11-16 21:26:00得到结果

1573910760000===	
    GdsSales{areaId='beijing', gdsId='gdsId03', amount=300.0}	
    GdsSales{areaId='beijing', gdsId='gdsId01', amount=400.0}	
    GdsSales{areaId='beijing', gdsId='gdsId02', amount=500.0}

接着在生产一条数据

orderId02,1573874540000,gdsId04,500,beijing

等到2019-11-16 21:27:00得到结果

1573910820000===	
    GdsSales{areaId='beijing', gdsId='gdsId01', amount=400.0}	
    GdsSales{areaId='beijing', gdsId='gdsId02', amount=500.0}	
    GdsSales{areaId='beijing', gdsId='gdsId04', amount=700.0}

至此完成全局topN的全部实现。

### 总结 ###

全局TopN要求状态保存所有的聚合数据，对于key比较多的情况，不管是销售额数据还是定时器数据都会占用比较多的内存，可以选择RocksDb作为StateBackend。

-----------------------------------

精彩推文：

关注回复Flink获取系列文章~

![640?wx\_fmt=png][640_wx_fmt_png 1]

[640_wx_fmt_png]: https://ss.csdn.net/p?https://mmbiz.qpic.cn/mmbiz_png/S7tIvo5xYM4j97XRykdkDLIz5GZn7n7UK3zM3vZEgdVBACyVfd4PrAVTJic1EC9TjwLspvXDGRll4e0ABujstpw/640?wx_fmt=png
[640_wx_fmt_png 1]: https://ss.csdn.net/p?https://mmbiz.qpic.cn/mmbiz_png/S7tIvo5xYM57sWeXy2icLStv81CQSVH5AJkBxiaOYdOEiaicL3TujBgIbG3FYichgJWiavAhmEr78mAKXooxwvLTeSBA/640?wx_fmt=png

发表评论取消回复

表情：

评论列表（有 0 条评论，23人围观）

还没有评论，来说两句吧...

相关阅读

相关 Flink应用简单案例-统计TopN

?诉求：网站中一个非常经典的例子，就是实时统计一段时间内的热门 url。例如，需要统计最近10 秒钟内最热门的两个 url 链接，并且每 5 秒钟更新一次。我们知道，这可...

爱被打了一巴掌/ 2024年04月22日 07:01/ 0 赞/ 188 阅读

相关 Flink SQL TopN

文章目录 1、SQL 语法标准： 2、参数说明： 3、注意事项： Flink SQL 对于批处理（Batch）和流处理（streaming）模式

骑猪看日落/ 2024年03月23日 23:34/ 0 赞/ 225 阅读

相关大数据Flink（九十五）：DML：Window TopN

![cf73002ee5024ba2bf7bab1044bb0f97.jpeg][] 文章目录 DML：Window TopN --------------------

「爱情、让人受尽委屈。」/ 2024年02月28日 01:15/ 0 赞/ 151 阅读

相关大数据Flink（九十四）：DML：TopN 子句

![4d6d749d4a7e446d890c08e7c95461cd.jpeg][] 文章目录 DML：TopN 子句 -------------------- DM

比眉伴天荒/ 2024年02月28日 01:07/ 0 赞/ 153 阅读

相关 Flink实战:全局TopN分析与实现

点击蓝字 ![640?wx\_fmt=png][640_wx_fmt_png] 关注~~ 按照区域areaId+商品gdsId分组，计算每个分组的累计销售额

野性酷女/ 2023年06月13日 15:28/ 0 赞/ 24 阅读

相关 Flink实战: 窗口TopN分析与实现

点击上方蓝字关注~ TopN 的需求场景不管是在离线计算还是实时计算都是比较常见的，例如电商中计算热门销售商品、广告计算中点击数前N的广告、搜索中计算搜索

太过爱你忘了你带给我的痛/ 2023年06月12日 15:24/ 0 赞/ 206 阅读

相关 Flink sql实现topN聚合结果写入kafka （Flink 1.11.0版本）

先跟鸡哥打个广告，博客地址: [https://me.csdn.net/weixin\_47482194][https_me.csdn.net_weixin_47482194

秒速五厘米/ 2023年03月05日 03:25/ 0 赞/ 140 阅读

相关 Spark：对数据实现TopN

对数据实现TopN 一、准备数据 1.1 创建 topn.txt 1.2 把数据上传到hdfs 二、I

野性酷女/ 2022年10月16日 15:26/ 0 赞/ 336 阅读

相关 flink 全局窗口

import org.apache.flink.streaming.api.datastream.AllWindowedStream; import org.a

末蓝、/ 2022年09月11日 14:25/ 0 赞/ 320 阅读

相关【零基础学flink】flink实战：使用flink分析wiki log

本文将从头开始，从设置Flink项目到在Flink集群上运行流分析程序。 Wikipedia提供了一个IRC频道，其中记录了对Wiki的所有编辑。我们将在Flink中读取此通道

「爱情、让人受尽委屈。」/ 2022年02月14日 07:49/ 0 赞/ 342 阅读