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

骑猪看日落 2023-02-25 07:17 82阅读 0赞

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

  1. import org.apache.flink.api.common.state.{ ValueState, ValueStateDescriptor}
  2. import org.apache.flink.streaming.api.TimeCharacteristic
  3. import org.apache.flink.streaming.api.functions.KeyedProcessFunction
  4. import org.apache.flink.streaming.api.scala.{ OutputTag, StreamExecutionEnvironment}
  5. import org.apache.flink.streaming.api.scala._
  6. import org.apache.flink.util.Collector
  9. object OrderTimeOutWithoutCep { 
  11.   val orderTimeOutputTag = new OutputTag[OrderResult]("orderTimeout")
  13.   def main(args: Array[String]): Unit = { 
  14.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  15.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  16.     env.setParallelism(1)
  18.     //读取订单数据
  19.     val orderEventStream = env.socketTextStream("hadoop102", 7777)
  20.       .map(data => { 
  21.         val dataArray = data.split(",")
  22.         OrderEvent(dataArray(0).trim.toLong, dataArray(1).trim, dataArray(2).trim, dataArray(3).trim.toLong)
  23.       })
  24.       .assignAscendingTimestamps(_.eventTime * 1000L)
  25.       .keyBy(_.orderId)
  27.     //定义process function进行超时检测
  28.     val orderResultStream = orderEventStream
  29.       .process(new OrderPayMatch())
  31.     orderResultStream.print("payed")
  32.     orderResultStream.getSideOutput(orderTimeOutputTag).print("timeout")
  34.     env.execute("order timeout without cep job")
  36.   }
  38.   class OrderPayMatch() extends KeyedProcessFunction[Long, OrderEvent, OrderResult] { 
  40.     //保存pay是否来过的状态
  41.     lazy val isPayedState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean](
  42.       "ispayed-state", classOf[Boolean]))
  43.     //保存定时器的时间戳为状态
  44.     lazy val timeState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long](
  45.       "timer-state", classOf[Long]))
  47.     override def processElement(i: OrderEvent, context: KeyedProcessFunction[Long, OrderEvent, OrderResult]#Context, collector: Collector[OrderResult]): Unit = { 
  48.       //先读取状态
  49.       val isPayed = isPayedState.value()
  50.       val timerTs = timeState.value()
  52.       //根据事件的类型进行分类判断，做不同的处理逻辑
  53.       if (i.eventType == "create") { 
  54.         //1.如果是create事件，接下来判断pay是否来过
  55.         if ( isPayed ) { 
  56.           //1.1如过已经pay过，匹配成功，输出主流，清空状态
  57.           collector.collect(OrderResult(i.orderId, "payed successfully"))
  58.           context.timerService().deleteEventTimeTimer(timerTs)
  59.           isPayedState.clear()
  60.           timeState.clear()
  61.         } else { 
  62.           //如果没有pay过，注册定时器等待pay的到来
  63.           val ts = i.eventTime * 1000L + 15 * 60 * 1000L
  64.           context.timerService().registerEventTimeTimer(ts)
  65.           timeState.update(ts)
  66.         }
  67.       } else if (i.eventType == "pay") { 
  68.         //2.如果是pay事件，那么判断是否create过，用timer表示
  69.         if (timerTs > 0) { 
  70.           //2.1如果有定时器，说明已经有create来过
  71.           //继续判断，是否超过了timeout时间
  72.           if (timerTs > i.eventTime * 1000L) { 
  73.             //2.1.1如果定时器时间还没到，那么输出成功匹配 因为是pay先出现 然而前面create加15s 所以这里不加
  74.             collector.collect(OrderResult(i.orderId, "payed successfully"))
  75.           } else { 
  76.             //2.1.2如果当前pay的时间已经超时，那么输出到测输出流
  77.             context.output(orderTimeOutputTag, OrderResult(i.orderId, "payed but already timeout"))
  78.           }
  79.           //输出结束，清空状态
  80.           context.timerService().deleteEventTimeTimer(timerTs)
  81.           isPayedState.clear()
  82.           timeState.clear()
  83.         } else { 
  84.           //2.2pay先到了，更新状态，注册定时器等待create 乱序情况
  85.           //定时器时间戳：等到水位到了watermark=value.event*1000L 时看create来没来，如果没来则create丢失，因为create应该在pay之前
  86.           isPayedState.update(true)
  87.           context.timerService().registerEventTimeTimer(i.eventTime * 1000L)
  88.           timeState.update(i.eventTime * 1000L)
  89.         }
  90.       }
  91.     }
  93.     override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#OnTimerContext, out: Collector[OrderResult]): Unit = { 
  94.       //根据状态的值，判断哪个数据没来
  95.       if (isPayedState.value()) { 
  96.         //如果为true 表示pay先到，没等到create
  97.         ctx.output(orderTimeOutputTag, OrderResult(ctx.getCurrentKey, "already payed but not found create log"))
  98.       } else { 
  99.         //表示create到了，没等到pay
  100.         ctx.output(orderTimeOutputTag, OrderResult(ctx.getCurrentKey, "order timeout"))
  101.       }
  102.       isPayedState.clear()
  103.       timeState.clear()
  104.     }
  105.   }
  106. }

在这里插入图片描述
在这里插入图片描述
犯了一个低级错误，导致花了很长时间找错误
写成了

  1. orderEventStream.getSideOutput(orderTimeOutputTag).print("timeout")

正确是

  1. orderResultStream.getSideOutput(orderTimeOutputTag).print("timeout")

发表评论

表情：
评论列表 （有 0 条评论，82人围观）

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

相关阅读