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

淡淡的烟草味﹌ 2022-11-22 00:13 215阅读 0赞

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

import java.util
    import org.apache.flink.cep.{ PatternSelectFunction, PatternTimeoutFunction}
    import org.apache.flink.cep.scala.CEP
    import org.apache.flink.cep.scala.pattern.Pattern
    import org.apache.flink.streaming.api.TimeCharacteristic
    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
    
    //定义输入输出的样例类
    case class OrderEvent(orderId:Long,eventType:String,txId:String,eventTime:Long)
    case class OrderResult(orderId:Long,resultMsg:String)
    
    object OrderTimeOut { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        env.setParallelism(1)
    
        //从文件中读取数据，并转换成样例类
        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
          })
    
        //1.定义一个匹配事件序列的模式
        val orderPayPattern = Pattern
          .begin[OrderEvent]("create").where(_.eventType == "create")  //首先是订单的create事件
          .followedBy("pay").where(_.eventType == "pay") //后面来的是订单的pay事件
          .within(Time.minutes(15))
    
        //2.将pattern应用在按照orderId分组的数据流上
        val patterStream = CEP.pattern(orderEventStream.keyBy(_.orderId), orderPayPattern)
    
        //3.定义一个测输出流标签，用来标明超时事件的测输出流
        val orderTimeOutputTag = new OutputTag[OrderResult]("order time out")
    
        //4.调用select方法，提取匹配事件和超时事件，分别进行处理转换输出
        val result = patterStream
          .select(orderTimeOutputTag, new OrderTimeOutSelect(), new OrderPaySelect())
    
        //5.打印输出
        result.print("payed")
        result.getSideOutput(orderTimeOutputTag).print("timeout")
    
        env.execute("order timeout detect job")
      }
    }
    //自定义超时处理函数
    class OrderTimeOutSelect() extends PatternTimeoutFunction[OrderEvent,OrderResult]{ 
      override def timeout(pattern: util.Map[String, util.List[OrderEvent]], timeoutTimestamp: Long): OrderResult = { 
        val timeOutOrderId = pattern.get("create").iterator().next().orderId
        OrderResult(timeOutOrderId,"timeout at"+timeoutTimestamp)
      }
    }
    
    //自定义匹配处理函数
    class OrderPaySelect() extends PatternSelectFunction[OrderEvent,OrderResult]{ 
      override def select(pattern: util.Map[String, util.List[OrderEvent]]): OrderResult = { 
        val payedOrderId = pattern.get("pay").get(0).orderId
        OrderResult(payedOrderId,"pay successfully")
      }
    }

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/305bd1575e0d4b22b6758984eb46b05a.png