Flink操练（三十三）之自定义键控状态（三）MapState

　　0 简介
　　MapState[K, V]保存Key-Value对。 MapState.get(key: K) MapState.put(key: K, value: V) MapState.contains(key: K) MapState.remove(key: K) 1 实例
　　1.1 实例一：
　　去重计算应该是数据分析业务里面常见的指标计算，例如网站一天的访问用户数、广告的点击用户数等等，离线计算是一个全量、一次性计算的过程通常可以通过distinct的方式得到去重结果，而实时计算是一种增量、长期计算过程，我们在面对不同的场景，例如数据量的大小、计算结果精准度要求等可以使用不同的方案。
　　此篇介绍如何通过编码方式实现精确去重，以一个实际场景为例：计算每个广告每小时的点击用户数，广告点击日志包含：广告位ID、用户设备ID(idfa/imei/cookie)、点击时间。
　　实现步骤分析： 为了当天的数据可重现，这里选择事件时间也就是广告点击时间作为每小时的窗口期划分 数据分组使用广告位ID+点击事件所属的小时 选择processFunction来实现，一个状态用来保存数据、另外一个状态用来保存对应的数据量 计算完成之后的数据清理，按照时间进度注册定时器清理
　　广告数据 case class AdData(id:Int,devId:String,time:Long)
　　分组数据 case class AdKey(id:Int,time:Long)
　　主流程 val env=StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)  val kafkaConfig=new Properties() kafkaConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,＂localhost:9092＂) kafkaConfig.put(ConsumerConfig.GROUP_ID_CONFIG,＂test1＂)  val consumer=new FlinkKafkaConsumer[String](＂topic1＂,new SimpleStringSchema,kafkaConfig) val ds=env.addSource(consumer)         .map(x=>{                 val s=x.split(＂,＂)                 AdData(s(0).toInt,s(1),s(2).toLong)                 }             ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[AdData](Time.minutes(1)){                 override def extractTimestamp(element: AdData): Long = element.time             })       .keyBy(x=>{             val endTime=              TimeWindow.getWindowStartWithOffset(x.time, 0, Time.hours(1).toMilliseconds)+Time.hours(1).toMilliseconds             AdKey(x.id,endTime)       })
　　指定时间时间属性，这里设置允许1min的延时，可根据实际情况调整；
　　时间的转换选择TimeWindow.getWindowStartWithOffset Flink在处理window中自带的方法，使用起来很方便，第一个参数 表示数据时间，第二个参数offset偏移量，默认为0，正常窗口划分都是整点方式，例如从0开始划分，这个offset就是相对于0的偏移量，第三个参数表示窗口大小，得到的结果是数据时间所属窗口的开始时间，这里加上了窗口大小，使用结束时间与广告位ID作为分组的Key。
　　去重逻辑
　　自定义Distinct1ProcessFunction 继承了KeyedProcessFunction， 方便起见使用输出类型使用Void,这里直接使用打印控制台方式查看结果，在实际中可输出到下游做一个批量的处理然后在输出；
　　定义两个状态：MapState，key表示devId, value表示一个随意的值只是为了标识，该状态表示一个广告位在某个小时的设备数据，如果我们使用rocksdb作为statebackend, 那么会将mapstate中key作为rocksdb中key的一部分，mapstate中value作为rocksdb中的value, rocksdb中value 大小是有上限的，这种方式可以减少rocksdb value的大小；
　　另外一个ValueState,存储当前MapState的数据量，是由于mapstate只能通过迭代方式获得数据量大小，每次获取都需要进行迭代，这种方式可以避免每次迭代。 class Distinct1ProcessFunction extends KeyedProcessFunction[AdKey, AdData, Void] {    var devIdState: MapState[String, Int] = _   var devIdStateDesc: MapStateDescriptor[String, Int] = _   var countState: ValueState[Long] = _   var countStateDesc: ValueStateDescriptor[Long] = _      override def open(parameters: Configuration): Unit = {     devIdStateDesc = new MapStateDescriptor[String, Int](＂devIdState＂, TypeInformation.of(classOf[String]), TypeInformation.of(classOf[Int]))     devIdState = getRuntimeContext.getMapState(devIdStateDesc)     countStateDesc = new ValueStateDescriptor[Long](＂countState＂, TypeInformation.of(classOf[Long]))     countState = getRuntimeContext.getState(countStateDesc)   }    override def processElement(value: AdData, ctx: KeyedProcessFunction[AdKey, AdData, Void]#Context, out: Collector[Void]): Unit = {     val currW=ctx.timerService().currentWatermark()     if(ctx.getCurrentKey.time+1<=currW) {         println(＂late data:＂ + value)         return       }      val devId = value.devId     devIdState.get(devId) match {       case 1 => {         //表示已经存在       }        case _ => {         //表示不存在         devIdState.put(devId, 1)         val c = countState.value()         countState.update(c + 1)         //还需要注册一个定时器         ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey.time + 1)       }     }     println(countState.value())   }    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[AdKey, AdData, Void]#OnTimerContext, out: Collector[Void]): Unit = {     println(timestamp + ＂ exec clean~~~＂)     println(countState.value())     devIdState.clear()     countState.clear()   } }
　　数据清理通过注册定时器方式ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey.time + 1)表示当watermark大于该小时结束时间+1就会执行清理动作，调用onTimer方法。
　　在处理逻辑里面加了 val currW=ctx.timerService().currentWatermark() if(ctx.getCurrentKey.time+1<=currW){         println(＂late data:＂ + value)         return   }
　　主要考虑可能会存在滞后的数据比较严重，会影响之前的计算结果，做了一个类似window机制里面的一个延时判断，将延时的数据过滤掉，也可以使用OutputTag 单独处理。
　　1.2 实例二：
　　我们知道电商平台会将用户与商品的交互行为收集记录下来，行为数据主要包括几个字段：userId、itemId、categoryId、behavior和timestamp。其中userId和itemId分别代表用户和商品的唯一ID，categoryId为商品类目ID，behavior表示用户的行为类型，包括点击(pv)、购买(buy)、加购物车(cart)、喜欢(fav)等，timestamp记录行为发生时间。本文采用阿里巴巴提供的一个淘宝用户行为数据集，为了精简需要，只节选了部分数据。下面的代码使用 MapState[String, Int]  记录某个用户某种行为出现的次数。这里读取了数据集文件，模拟了一个淘宝用户行为数据流。/**   * 用户行为   * categoryId为商品类目ID   * behavior包括点击（pv）、购买（buy）、加购物车（cart）、喜欢（fav）   * */ case class UserBehavior(userId: Long,                           itemId: Long,                           categoryId: Int,                           behavior: String,                           timestamp: Long)  class MapStateFunction extends RichFlatMapFunction[UserBehavior, (Long, String, Int)] {    // 指向MapState的句柄   private var behaviorMapState: MapState[String, Int] = _    override def open(parameters: Configuration): Unit = {     // 创建StateDescriptor     val behaviorMapStateDescriptor = new MapStateDescriptor[String, Int](＂behaviorMap＂, classOf[String], classOf[Int])     // 通过StateDescriptor获取运行时上下文中的状态     behaviorMapState = getRuntimeContext.getMapState(behaviorMapStateDescriptor)   }    override def flatMap(input: UserBehavior, collector: Collector[(Long, String, Int)]): Unit = {     var behaviorCnt = 1     // behavior有可能为pv、cart、fav、buy等     // 判断状态中是否有该behavior     if (behaviorMapState.contains(input.behavior)) {       behaviorCnt = behaviorMapState.get(input.behavior) + 1     }     // 更新状态     behaviorMapState.put(input.behavior, behaviorCnt)     collector.collect((input.userId, input.behavior, behaviorCnt))   } }  def main(args: Array[String]): Unit = {    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment   env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)   env.setParallelism(8)    // 获取数据源   val sourceStream: DataStream[UserBehavior] = env   .addSource(new UserBehaviorSource(＂state/UserBehavior-50.csv＂)).assignTimestampsAndWatermarks(new AscendingTimestampExtractor[UserBehavior]() {     override def extractAscendingTimestamp(userBehavior: UserBehavior): Long = {       // 原始数据单位为秒，乘以1000转换成毫秒       userBehavior.timestamp * 1000     }   }                                                                                            )    // 生成一个KeyedStream   val keyedStream =  sourceStream.keyBy(user => user.userId)    // 在KeyedStream上进行flatMap   val behaviorCountStream = keyedStream.flatMap(new MapStateFunction)    behaviorCountStream.print()    env.execute(＂state example＂) }  class UserBehaviorSource(path: String) extends RichSourceFunction[UserBehavior] {    var isRunning: Boolean = true   // 输入源   var streamSource: InputStream = _    override def run(sourceContext: SourceContext[UserBehavior]): Unit = {     // 从项目的resources目录获取输入     streamSource = MapStateExample.getClass.getClassLoader.getResourceAsStream(path)     val lines: Iterator[String] = scala.io.Source.fromInputStream(streamSource).getLines     while (isRunning && lines.hasNext) {       val line = lines.next()       val itemStrArr = line.split(＂,＂)       val userBehavior = UserBehavior(itemStrArr(0).toLong, itemStrArr(1).toLong, itemStrArr(2).toInt, itemStrArr(3), itemStrArr(4).toLong)       sourceContext.collect(userBehavior)     }   }    override def cancel(): Unit = {     streamSource.close()     isRunning = false   } }
　　Keyed State是针对 KeyedStream  的状态，必须先对一个DataStream  进行keyBy  操作。在本例中，我们对用户ID进行了keyBy  ，那么用户ID为1的行为数据共享同一状态数据，以此类推，每个用户ID的行为数据共享自己的状态数据。
　　之后，我们需要实现Rich类函数，比如 RichFlatMapFunction  ，或者KeyedProcessFunction  等函数类。这些算子函数类都是RichFunction  的一种实现，他们都有运行时上下文RuntimeContext  ，RuntimeContext  包含了状态数据。 在实现这些算子函数类时，一般是在open  方法中声明状态。open  是算子的初始化方法，它在实际处理函数之前调用。
　　具体到状态的使用，我们首先要注册一个 StateDescriptor  。从名字中可以看出，StateDescriptor  是状态的一种描述，它描述了状态的名字和状态的数据结构。状态的名字可以用来区分不同的状态，一个算子内可以有多个不同的状态，每个状态的StateDescriptor  需要设置不同的名字。同时，我们也需要指定状态的具体数据结构，指定具体的数据结构非常重要，因为Flink要对其进行序列化和反序列化，以便进行Checkpoint和必要的恢复。数据结构的类型和序列化机制可以参考我之前的文章：Flink进阶教程：数据类型和序列化机制简介。
　　在本例中，我们使用 val behaviorMapStateDescriptor = new MapStateDescriptor[String, Int](＂behaviorMap＂, classOf[String], classOf[Int])  注册了一个MapStateStateDescriptor  ，key为某种行为，如pv、buy等，数据类型为String  ，value为该行为出现的次数，数据类型为Int  。此外，每种类型的状态都有对应的StateDescriptor  ，比如MapStateDescriptor  对应MapState  ，ValueStateDescriptor  对应ValueState  。
　　接着我们通过 StateDescriptor  向RuntimeContext  中获取状态句柄。本例中对应的代码为：behaviorMapState = getRuntimeContext.getMapState(behaviorMapStateDescriptor)  。状态句柄并不存储状态，它只是Flink提供的一种访问状态的接口，状态数据实际存储在State Backend中。
　　使用和更新状态发生在实际的处理函数上，比如 RichFlatMapFunction  中的flatMap  方法，在实现自己的业务逻辑时访问和修改状态，比如通过get  方法获取状态。
　　1.3 实例三 package qiuhua; import com.google.common.collect.Lists; import org.apache.flink.api.common.functions.RichFlatMapFunction; import org.apache.flink.api.common.state.MapState; import org.apache.flink.api.common.state.MapStateDescriptor; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.util.Collector; import java.util.ArrayList; import java.util.UUID; /**  * @program: bigdata_learn  * @description: 如果当前的 key 出现了 3 次，则需要计算平均值  * @author: Mr.逗  * @create: 2021-09-08 16:47  *  MapState ：这个状态为每一个 key 保存一个 Map 集合  *      put() 将对应的 key 的键值对放到状态中  *      values() 拿到 MapState 中所有的 value  *      clear() 清除状态  **/ public class CountAverageWithMapState extends RichFlatMapFunction,Tuple2> {     //1. MapState ：key 是一个唯一的值，value 是接收到的相同的 key 对应的 value 的值      @Override     public void open(Configuration parameters) throws Exception {         super.open(parameters);         //注册状态         MapStateDescriptor descriptor = new MapStateDescriptor<>(＂map_state＂, String.class, Long.class);         mapState=getRuntimeContext().getMapState(descriptor);     }      /**      * MapState:      *      Map集合的特点，相同key，会覆盖数据。      */      private MapState mapState;     @Override     public void flatMap(Tuple2 value, Collector> out) throws Exception {      mapState.put(UUID.randomUUID().toString(),value.f1);      //如果当前key出现了三次，就进行计算平均值         ArrayList allElements = Lists.newArrayList(mapState.values());         if (allElements.size()==3)         {             long count=0;             long sum=0;             for(Long ele:allElements)             {                 count++;                 sum+=ele;             }             double avg=(double)sum/count;             out.collect(Tuple2.of(value.f0,avg));             //清除状态             mapState.clear();         }     } }