Flink-电商用户行为分析(订单支付实时监控-状态编程实现)_2

た入场券 2022-11-21 15:11 195阅读 0赞

**`数据`**  
链接：https://pan.baidu.com/s/1nSMh3JaNDW1SheQ5I4J0FQ  
提取码：e49w  
在电商平台中，最终创造收入和利润的是用户下单购买的环节；更具体一点，是用户真正完成支付动作的时候。用户下单的行为可以表明用户对商品的需求，但在现实中，并不是每次下单都会被用户立刻支付。当拖延一段时间后，用户支付的意愿会降低。所以为了让用户更有紧迫感从而提高支付转化率，同时也为了防范订单支付环节的安全风险，电商网站往往会对订单状态进行监控，设置一个失效时间（比如 15 分钟），如果下单后一段时间仍未付，订单就会被取消。  
**`需求`**  
用户支付,订单15分钟内,查看支付成功或超时的  
来一条处理一条，正常的放主流，不正常的放侧输出流，

import org.apache.flink.api.common.state.{ ValueState, ValueStateDescriptor}
    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._
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.util.Collector
    
    object OrderTimeoutWithOutCep { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        //从文件中读取数据，并转换成样例类
        val resource = getClass.getResource("/OrderLog.csv")
        val orderEventStream = env.readTextFile(resource.getPath)
          .map(data => { 
            val dataArray = data.split(",")
            OrderEvent(dataArray(0).toLong, dataArray(1), dataArray(2), dataArray(3).toLong)
          })
          .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderEvent](Time.seconds(3)) { 
            override def extractTimestamp(element: OrderEvent): Long = element.eventTime * 1000L
          })
    
        //自定义Process Funcition，做精细化的流程控制
        val orderResultStream = orderEventStream
          .keyBy(_.orderId)
          .process(new OrderPayMatchDetect())
    
        //打印输出
        orderResultStream.print("payed")
        orderResultStream.getSideOutput(new OutputTag[OrderResult]("timeout")).print("timeout")
    
        env.execute("order timeout without cep job")
    
      }
    }
    class OrderPayMatchDetect() extends KeyedProcessFunction[Long,OrderEvent,OrderResult]{ 
      //定义状态，用来保存是否来过create和pay事件的标识位，以及定时器的时间戳
      lazy val isPayState:ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("is-payed",
        classOf[Boolean]))
      lazy val isCreateState:ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]
      ("is-created", classOf[Boolean]))
      lazy val timerTsState:ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timer-ts",
        classOf[Long]))
      val orderTimeOutputTag = new OutputTag[OrderResult]("timeout")
    
      override def processElement(value: OrderEvent, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]
        #Context, out: Collector[OrderResult]): Unit = { 
        //先取出当前状态
        val isPayed = isPayState.value()
        val isCreated = isCreateState.value()
        val timeTs = timerTsState.value()
    
    
    
        //判断当前事件的类型，分成不同情况讨论
        //情况1：来的是create，要继续判断之前是否有pay来过
        if (value.eventType == "create"){ 
          //情况1.1：如果已经pay过的话，匹配成功,输出到主流,清空状态
          if (isPayed){ 
            out.collect(OrderResult(value.orderId,"payed successfully"))
            isPayState.clear()
            timerTsState.clear()
            ctx.timerService().deleteEventTimeTimer(timeTs)
          }
          //情况1.2：如果没有pay过，那么就注册一个15钟后的定时器，开始等待
          else{ 
            val ts = value.eventTime * 1000L + 15 * 60 *1000L
            ctx.timerService().registerEventTimeTimer(ts)
            timerTsState.update(ts)
            isCreateState.update(true)
          }
        }
        //情况2，来的是pay，要继续判断是否来过create
        else if (value.eventType == "pay"){ 
          //情况2.1：如果create已经来过，匹配成功，要继续判断间隔时间是否超过了15分钟
          if (isCreated){ 
            //情况2.1.1：如果没有超时，正常输出结果到主流
            if (value.eventTime * 1000L < timeTs){ 
              out.collect(OrderResult(value.orderId,"payed successfully"))
            }else{ 
              //情况2.1.2：如果已经超时，那么输出timeout报警到侧输出流
              ctx.output(orderTimeOutputTag,OrderResult(value.orderId,"payed but already timeout"))
            }
            //无论哪种情况，都已经有了输出，清空状态
            isCreateState.clear()
            timerTsState.clear()
            ctx.timerService().deleteEventTimeTimer(timeTs)
          }
          //情况2.2：如果create没来，需要等待乱序create，注册一个当前pay时间戳的定时器
          else{ 
            val ts = value.eventTime * 1000L
            ctx.timerService().registerEventTimeTimer(ts)
            timerTsState.update(ts)
            isPayState.update(true)
          }
        }
      }
      override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]
        #OnTimerContext, out: Collector[OrderResult]): Unit = { 
        //定时器触发，要判断是哪种情况
        if (isPayState.value()){ 
          //如果pay过，那么说明create没来，可能出现了数据丢失异常情况
          ctx.output(orderTimeOutputTag,OrderResult(ctx.getCurrentKey,"already payed but not found created log"))
        }else{ 
          //如果没有pay过，那么说明真正15分钟超时
          ctx.output(orderTimeOutputTag,OrderResult(ctx.getCurrentKey,"order timeout"))
        }
        //清空状态
        isPayState.clear()
        isCreateState.clear()
        timerTsState.clear()
      }
    }