Flink-处理乱序数据和迟到数据的总结和解决办法

雨点打透心脏的1/2处 2022-12-18 11:56 248阅读 0赞

Flink有三重保证

watermark可以设置延迟时间
    window的allowedLateness方法，可以设置窗口允许处理迟到数据的时间
    window的sideOutputLateData方法，可以将迟到的数据写入侧输出流

我以我之前做的项目进行优化  
[https://blog.csdn.net/qq\_46548855/article/details/107170533][https_blog.csdn.net_qq_46548855_article_details_107170533]  
看看乱序数据(大约最大时间相差50s，那么我设置成1分钟)  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]  
**`提醒：尽量不要watermark不要设置成60s，因为10:13.50的数据要等到10:14:50才输出，这样flink实时性太差了！把watermark设置成1s，把allowedLateness(Time.seconds(60))设置成60s`**  
我已经设置了watermark在29行和40行  
allowedLateness在50行  
sideOutputLateData在45，51，54行

import java.sql.Timestamp
    import java.text.SimpleDateFormat
    
    import org.apache.flink.api.common.functions.AggregateFunction
    import org.apache.flink.api.common.state.{ ListState, ListStateDescriptor}
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
    import org.apache.flink.streaming.api.scala.{ OutputTag, StreamExecutionEnvironment}
    import org.apache.flink.streaming.api.scala.function.WindowFunction
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    
    import scala.collection.mutable.ListBuffer
    
    //204.62.56.3 - - 20/05/2015:17:05:11 +0000 GET /presentations/logstash-puppetconf-2012/images/nagios-sms2.png
    //输入数据样例类
    case class ApacheLogEvent(ip:String,userId:String,evetTime:Long,method:String,url:String)
    
    //窗口聚合结果样例类
    case class UrlViewCount(url:String,windowEnd:Long,count:Long)
    
    object NetworkFlowTopNPage { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
       val inputStream = env.socketTextStream("hadoop12",9999)
        val dataStream = inputStream.map(data => { 
          val dataArray = data.split(" ")
          //定义事件转换 20/05/2015:17:05:11
          val simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
          val timestamp = simpleDateFormat.parse(dataArray(3).trim).getTime
          ApacheLogEvent(dataArray(0).trim, dataArray(1).trim, timestamp, dataArray(5).trim, dataArray(6).trim)
        })
          //*1000看是秒还是毫秒 秒*1000 毫秒不乘
          .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[ApacheLogEvent](Time.seconds(1)) { 
            override def extractTimestamp(t: ApacheLogEvent): Long = t.evetTime
          })
    
        //开窗聚合
        val lateOutpUTag = new OutputTag[ApacheLogEvent]("late data")
        val aggStream = dataStream
          .keyBy(_.url)
          .timeWindow(Time.minutes(10), Time.seconds(5))
          //允许60秒的延迟数据去更新
          .allowedLateness(Time.seconds(60))
          .sideOutputLateData(lateOutpUTag)
          .aggregate(new CountAgg(), new WindowResult())
    
        aggStream.getSideOutput(lateOutpUTag)
    
        val resultStream = aggStream
          .keyBy(_.windowEnd)
          .process(new TopNHotUrls(5))
    
        dataStream.print("data")
        aggStream.print("agg")
        lateDataStream.print("late")
        resultStream.print("result")
        
        env.execute("network flow job")
      }
    }
    //自定义预聚合函数
    class CountAgg() extends AggregateFunction[ApacheLogEvent,Long,Long]{ 
      override def createAccumulator(): Long = 0L
    
      override def add(in: ApacheLogEvent, acc: Long): Long = acc+1
    
      override def getResult(acc: Long): Long = acc
    
      override def merge(acc: Long, acc1: Long): Long = acc + acc1
    }
    //自定义窗口函数
    class WindowResult() extends WindowFunction[Long,UrlViewCount,String,TimeWindow] { 
      override def apply(key: String, window: TimeWindow, input: Iterable[Long],
                         out: Collector[UrlViewCount]): Unit = { 
        out.collect(UrlViewCount(key, window.getEnd, input.iterator.next()))
      }
    }
    
    //自定义的处理函数 第一个windowend的类型
    class TopNHotUrls(topSize: Int) extends KeyedProcessFunction[Long,UrlViewCount,String]{ 
    
      lazy val urlState:ListState[UrlViewCount] = getRuntimeContext.getListState(new ListStateDescriptor[UrlViewCount]
      ("url-state",classOf[UrlViewCount]))
    
      override def processElement(i: UrlViewCount, context: KeyedProcessFunction[Long, UrlViewCount, String]
        #Context, collector: Collector[String]): Unit = { 
        urlState.add(i)
        context.timerService().registerEventTimeTimer(i.windowEnd + 1)
      }
    
      override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, UrlViewCount, String]
        #OnTimerContext, out: Collector[String]): Unit = { 
        //从状态中拿到数据
        val allUrlViews:ListBuffer[UrlViewCount] = new ListBuffer[UrlViewCount]
        val iter = urlState.get().iterator()
        while (iter.hasNext){ 
          allUrlViews += iter.next()
        }
        urlState.clear()
        //sortWith从高到底
        val sortedUrlViews = allUrlViews.sortWith(_.count>_.count).take(topSize)
    
        //格式化输出
        val result:StringBuilder = new StringBuilder()
        result.append("时间:").append(new Timestamp(timestamp - 1)).append("\n")
        for (i <- sortedUrlViews.indices){ 
          val currentUrlView = sortedUrlViews(i)
          result.append("NO").append(i+1).append(":")
            .append("URL=").append(currentUrlView.url)
            .append("访问量=").append(currentUrlView.count).append("\n")
        }
        result.append("=====================")
        Thread.sleep(1000)
        out.collect(result.toString())
    
      }
    }

在虚拟机上输入，进行测试

nc -lk 9999

![在这里插入图片描述][20201026203045896.png_pic_center]  
为什么51秒的时候才输出，因为watermark设置成1，滑动步长为5s，因为要整数倍，才会触发watermark,所有agg在1秒之后才触发  
为啥result在52秒才能输出，因为设置的定时器在为1毫秒，所以result在1s+1ms之后才会输出(代码在97行)  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]  
发一条迟来数据

83.149.9.216 - - 17/05/2015:10:06:46 +0000 GET /presentations
    83.149.9.216 - - 17/05/2015:10:06:53 +0000 GET /presentations

因为46秒那个窗口进行聚合，然而并没有输出result，因为watermark没有更新  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]  
再继续测试

83.149.9.216 - - 17/05/2015:10:06:46 +0000 GET /present
    83.149.9.216 - - 17/05/2015:10:06:54 +0000 GET /presentations

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]  
发现了Bug，出新的url，结果输出了当前的top1，因为把状态清空了(代码在108行)  
**`因此我们还要注册一个定时器`**  
修改了第10行(因为注册定时器60s，因为我设置的延时时间1分钟，那么注册两个定时器，无法知道什么时候触发哪个定时器，所以我们需要进行判断)，16-19行(进行判断)

class TopNHotUrls(topSize: Int) extends KeyedProcessFunction[Long,UrlViewCount,String]{ 
    
      lazy val urlState:ListState[UrlViewCount] = getRuntimeContext.getListState(new ListStateDescriptor[UrlViewCount]
      ("url-state",classOf[UrlViewCount]))
    
      override def processElement(i: UrlViewCount, context: KeyedProcessFunction[Long, UrlViewCount, String]
        #Context, collector: Collector[String]): Unit = { 
        urlState.add(i)
        context.timerService().registerEventTimeTimer(i.windowEnd + 1)
        context.timerService().registerEventTimeTimer(i.windowEnd + 60 * 1000L)
      }
    
      override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, UrlViewCount, String]
        #OnTimerContext, out: Collector[String]): Unit = { 
        //ctx.getCurrentKey就时当前的windowEnd
        if (timestamp == ctx.getCurrentKey + 60000L){ 
          urlState.clear()
          return
        }
    
        //从状态中拿到数据
        val allUrlViews:ListBuffer[UrlViewCount] = new ListBuffer[UrlViewCount]
        val iter = urlState.get().iterator()
        while (iter.hasNext){ 
          allUrlViews += iter.next()
        }
    
        //urlState.clear()
    
        //sortWith从高到底
        val sortedUrlViews = allUrlViews.sortWith(_.count>_.count).take(topSize)
    
        //格式化输出
        val result:StringBuilder = new StringBuilder()
        result.append("时间:").append(new Timestamp(timestamp - 1)).append("\n")
        for (i <- sortedUrlViews.indices){ 
          val currentUrlView = sortedUrlViews(i)
          result.append("NO").append(i+1).append(":")
            .append("URL=").append(currentUrlView.url)
            .append("访问量=").append(currentUrlView.count).append("\n")
        }
        result.append("=====================")
        Thread.sleep(1000)
        out.collect(result.toString())
    
      }
    }

我们再进行测试

83.149.9.216 - - 17/05/2015:10:06:49 +0000 GET /presentations
    83.149.9.216 - - 17/05/2015:10:06:50 +0000 GET /presentations
    83.149.9.216 - - 17/05/2015:10:06:51 +0000 GET /presentations
    83.149.9.216 - - 17/05/2015:10:06:52 +0000 GET /presentations
    83.149.9.216 - - 17/05/2015:10:06:46 +0000 GET /presentations
    83.149.9.216 - - 17/05/2015:10:06:53 +0000 GET /presentations

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 4]  
发现还是有Bug，因为ListState不能更新，只能追加，那么我们可以使用MapState

//自定义的处理函数 第一个windowend的类型
    class TopNHotUrls(topSize: Int) extends KeyedProcessFunction[Long,UrlViewCount,String]{ 
    
      //定义MapState
      lazy val urlState:MapState[String,Long] = getRuntimeContext.getMapState(new MapStateDescriptor[String,Long]
      ("url-state",classOf[String],classOf[Long]))
    
      override def processElement(i: UrlViewCount, context: KeyedProcessFunction[Long, UrlViewCount, String]
        #Context, collector: Collector[String]): Unit = { 
        urlState.put(i.url,i.count)
        context.timerService().registerEventTimeTimer(i.windowEnd + 1)
        context.timerService().registerEventTimeTimer(i.windowEnd + 60 * 1000L)
      }
    
      override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, UrlViewCount, String]
        #OnTimerContext, out: Collector[String]): Unit = { 
        //ctx.getCurrentKey就时当前的windowEnd
        if (timestamp == ctx.getCurrentKey + 60000L){ 
          urlState.clear()
          return
        }
    
        //从状态中拿到数据
        val allUrlViews:ListBuffer[(String,Long)] = ListBuffer()
        val iter = urlState.entries().iterator()
        while (iter.hasNext){ 
          val entry = iter.next()
          allUrlViews += ((entry.getKey,entry.getValue))
        }
    
        //urlState.clear()
    
        //sortWith从高到底
        val sortedUrlViews = allUrlViews.sortWith(_._2>_._2).take(topSize)
    
        //格式化输出
        val result:StringBuilder = new StringBuilder()
        result.append("时间:").append(new Timestamp(timestamp - 1)).append("\n")
        for (i <- sortedUrlViews.indices){ 
          val currentUrlView = sortedUrlViews(i)
          result.append("NO").append(i+1).append(":")
            .append("URL=").append(currentUrlView._1)
            .append("访问量=").append(currentUrlView._2).append("\n")
        }
        result.append("=====================")
        Thread.sleep(1000)
        out.collect(result.toString())
    
      }
    }

继续进行测试

这下没问题了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 5]  
再出一个问题：侧输出流什么时候输出？  
十分钟的窗口，然后我们延迟1分钟那么就是11分钟，最后输出时间10:06:52，侧输出流是输出不在窗口之内的，那么窗口\[09:56:52,10:06:53),窗口是左闭右开，09:56:51就会输出测输出流

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 6]

[https_blog.csdn.net_qq_46548855_article_details_107170533]: https://blog.csdn.net/qq_46548855/article/details/107170533
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221122/c3cbea46e1d94814a5f0d7bfe1696795.png
[20201026203045896.png_pic_center]: /images/20221122/55c624e9f2224c7e8fde65d70c6426c0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221122/d2e413f3289b42d4a406f4c74dfcee9b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221122/d5b45a8c212d4c8fabc7d7681cef6f6e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221122/52b39f1fd79e4c5c936c211756cee97e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221122/44d290f1a0d34ffd962ed2e79000d343.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221122/01f03bcf07f3401ea4623104eb4c1abb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221122/b74ec8cee0bc46b4b06acdb9cf2db5da.png