基于 flink 的電商用戶行為資料分析【8】| 訂單支付實時監控-有解無憂

本篇是flink 的「電商用戶行為資料分析」的第 8 篇文章，為大家帶來的是市場營銷商業指標統計分析之訂單支付實時監控的內容！通過本期內容，我們可以實作通過使用CEP和Process Function來實作訂單支付實時監控的功能，還能學會通過connect 和 join來實作flink雙流join的功能，可謂干貨滿滿！受益的朋友記得三連支持一下 ~

在這里插入圖片描述

訂單支付實時監控

在電商網站中，訂單的支付作為直接與營銷收入掛鉤的一環，在業務流程中非常重要，對于訂單而言，為了正確控制業務流程，也為了增加用戶的支付意愿，網站一般會設定一個支付失效時間，超過一段時間不支付的訂單就會被取消，另外，對于訂單的支付，我們還應保證用戶支付的正確性，這可以通過第三方支付平臺的交易資料來做一個實時對賬，在接下來的內容中，我們將實作這兩個需求，

模塊創建和資料準備

同樣地，在UserBehaviorAnalysis下新建一個 maven module作為子專案，命名為OrderTimeoutDetect，在這個子模塊中，我們同樣將會用到 flink 的 CEP 庫來實作事件流的模式匹配，所以需要在pom檔案中引入CEP的相關依賴：

<dependency>
        <groupId>org.apache.flink</groupId>
<artifactId>flink-cep-scala_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>

同樣，在src/main/目錄下，將默認源檔案目錄java改名為scala，

代碼實作

在電商平臺中，最終創造收入和利潤的是用戶下單購買的環節；更具體一點，是用戶真正完成支付動作的時候，用戶下單的行為可以表明用戶對商品的需求，但在現實中，并不是每次下單都會被用戶立刻支付，當拖延一段時間后，用戶支付的意愿會降低，所以為了讓用戶更有緊迫感從而提高支付轉化率，同時也為了防范訂單支付環節的安全風險，電商網站往往會對訂單狀態進行監控，設定一個失效時間（比如15分鐘），如果下單后一段時間仍未支付，訂單就會被取消，

使用CEP實作

我們首先還是利用CEP庫來實作這個功能，我們先將事件流按照訂單號orderId分流，然后定義這樣的一個事件模式：在15分鐘內，事件“create”與“pay”非嚴格緊鄰：

    // 1、 定義一個匹配事件序列的模式
    val orderPayPattern = Pattern
      .begin[OrderEvent]("create").where(_.eventType == "create")   // 首先是訂單的 create 事件
      .followedBy("pay").where(_.eventType == "pay")  // 后面來的是訂單的 pay 事件
      .within(Time.minutes(15))    // 間隔 15 分鐘

這樣呼叫.select方法時，就可以同時獲取到匹配出的事件和超時未匹配的事件了，

在src/main/scala下繼續創建OrderTimeout.scala檔案，新建一個單例物件，定義樣例類OrderEvent，這是輸入的訂單事件流；另外還有OrderResult，這是輸出顯示的訂單狀態結果，訂單資料也本應該從UserBehavior日志里提取，由于UserBehavior.csv中沒有做相關埋點，我們從另一個檔案OrderLog.csv中讀取登錄資料，

OrderLog.csv部分資料
完整代碼如下：

import java.util

import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.{PatternSelectFunction, PatternTimeoutFunction}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
/*
 * @Author: Alice菌
 * @Date: 2020/12/13 15:46
 * @Description:

 */
object OrderTimeoutWithOutCep {
   
  // 定義輸入的訂單事件樣例類
  case class OrderEvent(orderId:Long,eventType:String,eventTime:Long)
  // 定義輸出的訂單檢測結果樣例類
  case class OrderResult(orderId:Long,resultMsg:String)

  def main(args: Array[String]): Unit = {

    // 定義流處理環境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 設定程式并行度
    env.setParallelism(1)
    // 設定時間特征為事件時間
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 從檔案中讀取資料，并轉換成樣例類
    val orderEventStream: DataStream[OrderEvent] = env.readTextFile("YOUR_PATH\\OrderLog.csv")
      .map(data => {
        // 樣例資料： 34729,pay,sd76f87d6,1558430844
        val dataArray: Array[String] = data.split(",")
        OrderEvent(dataArray(0).toLong, dataArray(1), 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))    // 間隔 15 分鐘

    // 2、 將 pattern 應用在按照 orderId分組的資料流上
    val patterStream: PatternStream[OrderEvent] = CEP.pattern(orderEventStream.keyBy(_.orderId), orderPayPattern)

    // 3、定義一個側輸出流標簽，用來標明超時事件的側輸出流
    val orderTimeOutputTag: OutputTag[OrderResult] = new OutputTag[OrderResult]("order time out")

    // 4、呼叫select方法，提取匹配事件和超時事件，分別進行處理轉換輸出
    val result: DataStream[OrderResult] = 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: Long = 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: Long = pattern.get("pay").get(0).orderId
      OrderResult(payedOrderId,"pay successfully")
    }
  }
}

運行結果：

部分結果展示

使用Process Function實作

我們同樣可以利用Process Function，自定義實作檢測訂單超時的功能，為了簡化問題，我們只考慮超時報警的情形，在pay事件超時未發生的情況下，輸出超時報警資訊，

一個簡單的思路是，可以在訂單的 create 事件到來后注冊定時器，15分鐘后觸發；然后再用一個布爾型別的Value狀態來作為標識位，表明pay事件是否發生過，如果pay事件已經發生，狀態被置為true，那么就不再需要做什么操作；而如果pay事件一直沒來，狀態一直為false，到定時器觸發時，就應該輸出超時報警資訊，

具體代碼實作如下：

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


/*
 * @Author: Alice菌
 * @Date: 2020/12/23 19:35
 * @Description: 
    
 */
object OrderTimeout {

  // 定義輸入的訂單事件樣例類
  case class OrderEvent(orderId: Long, eventType: String, eventTime: Long)

  // 定義輸出的訂單檢測結果樣例類
  case class OrderResult(orderId: Long, resultMsg: String)

  def main(args: Array[String]): Unit = {
    // 定義流處理環境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 設定程式并行度
    env.setParallelism(1)
    // 設定時間特征為事件時間
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 讀取輸入的訂單資料流
    val orderEventStream: DataStream[OrderEvent] = env.readTextFile("YOUR_PATH\\OrderLog.csv")
      .map(data => {
        // 示例資料： 34729,pay,sd76f87d6,1558430844
        val dataArray: Array[String] = data.split(",")
        OrderEvent(dataArray(0).toLong, dataArray(1), dataArray(3).toLong)
      })
      // 設定水印
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderEvent](Time.seconds(3)) {
        override def extractTimestamp(element: OrderEvent): Long = element.eventTime * 1000L
      })

    // 自定義 Process Function,做精細化的流程控制
    val orderResultStream: DataStream[OrderResult] = 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: Boolean = isPayState.value()
      val isCreated: Boolean = isCreateState.value()
      val timeTs: Long = 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: Long = value.eventTime * 1000L + 15 * 60 *1000L
          // 設定一個15分鐘的定時器
          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: Long = 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 分鐘 超時 [提交了訂單，但是超過了15分鐘仍未支付]
        ctx.output(orderTimeOutputTag,OrderResult(ctx.getCurrentKey,"order timeout"))
      }

      // 清空狀態
      isPayState.clear()
      isCreateState.clear()
      timerTsState.clear()

    }
  }
}

運行結果：

部分結果展示

來自兩條流的訂單交易匹配

對于訂單支付事件，用戶支付完成其實并不算完，我們還得確認平臺賬戶上是否到賬了，而往往這會來自不同的日志資訊，所以我們要同時讀入兩條流的資料來做合并處理，這里我們利用connect將兩條流進行連接，然后用自定義的CoProcessFunction進行處理，

具體代碼如下：

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
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

/*
 * @Author: Alice菌
 * @Date: 2020/12/13 15:57
 * @Description: 
    來自兩條流的訂單交易匹配  （ connect 實作 ）
 */
object OrderPayTxMatch {

  // 輸入輸出的樣例類
  case class ReceiptEvent(txId:String, payChannel:String, timestamp:Long)
  case class OrderEvent(orderId:Long, eventType:String, txId:String, eventTime:Long)

  def main(args: Array[String]): Unit = {
    // 創建流處理的環境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 設定程式并行度
    env.setParallelism(1)
    // 設定時間特征為事件時間
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 從 OrderLog.csv 檔案中讀取資料 ，并轉換成樣例類
    val orderEventStream: KeyedStream[OrderEvent, String] = env.readTextFile("G:\\idea arc\\BIGDATA\\project\\src\\main\\resources\\OrderLog.csv")
      .map(data => {
        // 樣例資料 ：  34731,pay,35jue34we,1558430849
        val dataArray: Array[String] = 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
      })       //  為資料流中的元素分配時間戳
      .filter(_.eventType != "") // 只過濾出pay事件
      .keyBy(_.txId)    // 根據 訂單id 分組

    // 從 ReceiptLog.csv 檔案中讀取資料 ，并轉換成樣例類
    val receiptStream: KeyedStream[ReceiptEvent, String] = env.readTextFile("YOUR_PATH\\ReceiptLog.csv")
      .map(data => {
        // 樣例資料： 3hu3k2432,alipay,1558430848
        val dataArray: Array[String] = data.split(",")
        ReceiptEvent(dataArray(0), dataArray(1), dataArray(2).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L) // 設定水印
      .keyBy(_.txId)   // 根據 txId 進行分組

    // connect 連接兩條流，匹配事件進行處理
    val resultStream: DataStream[(OrderEvent, ReceiptEvent)] = orderEventStream.connect(receiptStream)
      .process(new OrderPayTxDetect())

    // 定義側輸出流
    val unmatchedPays: OutputTag[OrderEvent] = new OutputTag[OrderEvent]("unmatched-pays")
    val unmatchedReceipts: OutputTag[ReceiptEvent] = new OutputTag[ReceiptEvent]("unmatched-receipts")

    // 列印輸出
    resultStream.print("matched")
    resultStream.getSideOutput(unmatchedPays).print("unmatched-pays")
    resultStream.getSideOutput(unmatchedReceipts).print("unmatched-receipts")
    env.execute("order pay tx match job")

  }

  // 定義 CoProcessFunction，實作兩條流資料的匹配檢測
  class OrderPayTxDetect() extends CoProcessFunction[OrderEvent,ReceiptEvent,(OrderEvent,ReceiptEvent)]{

    // 定義兩個 ValueState，保存當前交易對應的支付事件和到賬事件
    lazy val payState: ValueState[OrderEvent] = getRuntimeContext.getState(new ValueStateDescriptor[OrderEvent]("pay", classOf[OrderEvent]))
    lazy val receiptState: ValueState[ReceiptEvent] = getRuntimeContext.getState(new ValueStateDescriptor[ReceiptEvent]("receipt", classOf[ReceiptEvent]))

    //定義側輸出流
    val unmatchedPays: OutputTag[OrderEvent] = new OutputTag[OrderEvent]("unmatched-pays")
    val unmatchedReceipts: OutputTag[ReceiptEvent] = new OutputTag[ReceiptEvent]("unmatched-receipts")

    override def processElement1(pay: OrderEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
      // pay 來了，考察有沒有對應的 receipt 來過
      val receipt: ReceiptEvent = receiptState.value()
      if (receipt != null){
        // 如果已經有 receipt，正常輸出到主流
        out.collect((pay,receipt))
        receiptState.clear()
      }else{
        // 如果 receipt 還沒來，那么把 pay 存入狀態，注冊一個定時器等待 5 秒
        payState.update(pay)
        ctx.timerService().registerEventTimeTimer(pay.eventTime * 1000L + 5000L)
      }
    }

    override def processElement2(receipt: ReceiptEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
      //receipt來了，考察有沒有對應的pay來過
      val pay: OrderEvent = payState.value()
      if (pay != null) {
        //如果已經有pay，那么正常匹配，輸出到主流
        out.collect((pay, receipt))
        payState.clear()
      }else{
      // 如果 pay 還沒來，那么把 receipt 存入狀態，注冊一個定時器等待 3 秒
        receiptState.update(receipt)
        ctx.timerService().registerEventTimeTimer(receipt.timestamp * 1000L + 3000L)
    }
  }

    // 定時觸發， 有兩種情況，所以要判斷當前有沒有pay和receipt
    override def onTimer(timestamp: Long, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]
      #OnTimerContext, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {

      //如果 pay 不為空，說明receipt沒來，輸出unmatchedPays
      if (payState.value() != null){
        ctx.output(unmatchedPays,payState.value())
      }

      if (receiptState.value() != null){
      ctx.output(unmatchedReceipts,receiptState.value())
      }

      // 清除狀態
      payState.clear()
      receiptState.clear()
    }
    }
}

運行結果：
部分結果展示
對于flink的雙流join通過connect的做法，肯定會有小伙伴覺得程序比較冗復雜，那還有沒有其他的方法也能實作類似的效果呢？
在這里插入圖片描述
當然是有的，下面就為大家展示另一種通過intervalJoin方法實作的方式：

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
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

/*
 * @Author: Alice菌
 * @Date: 2020/12/12 20:23
 * @Description: 
     來自兩條流的訂單交易匹配  （  JOIN 實作 ）
 */
object OrderPayTxMatchWithJoin {

  // 輸入輸出的樣例類
  case class ReceiptEvent(txId:String, payChannel:String, timestamp:Long)
  case class OrderEvent(orderId:Long, eventType:String, txId:String, eventTime:Long)

  def main(args: Array[String]): Unit = {
    // 創建流處理的環境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 設定程式并行度
    env.setParallelism(1)
    // 設定時間特征為事件時間
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 從 OrderLog.csv 檔案中讀取資料 ，并轉換成樣例類
    val orderEventStream: KeyedStream[OrderEvent, String] = env.readTextFile("YOUR_PATH\\OrderLog.csv")
      .map(data => {
        // 樣例資料 ：  34731,pay,35jue34we,1558430849
        val dataArray: Array[String] = 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
      })       //  為資料流中的元素分配時間戳
      .filter(_.eventType != "") // 只過濾出pay事件
      .keyBy(_.txId)    // 根據 訂單id 分組

    // 從 ReceiptLog.csv 檔案中讀取資料 ，并轉換成樣例類
    val receiptStream: KeyedStream[ReceiptEvent, String] = env.readTextFile("YOUR_PATH\\ReceiptLog.csv")
      .map(data => {
        // 樣例資料： 3hu3k2432,alipay,1558430848
        val dataArray: Array[String] = data.split(",")
        ReceiptEvent(dataArray(0), dataArray(1), dataArray(2).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L) // 設定水印
      .keyBy(_.txId)   // 根據 txId 進行分組

    // 使用 join 連接兩條流
    val resultStream: DataStream[(OrderEvent, ReceiptEvent)] = orderEventStream
      .intervalJoin(receiptStream)
      .between(Time.seconds(-5), Time.seconds(3))    
      .process(new OrderPayTxDetectWithJoin())

    resultStream.print()
    env.execute("order pay tx match with join job")

  }

  // 自定義 ProcessJoinFunction
  class OrderPayTxDetectWithJoin() extends ProcessJoinFunction[OrderEvent,ReceiptEvent,(OrderEvent,ReceiptEvent)]{
    override def processElement(left: OrderEvent, right: ReceiptEvent, ctx: ProcessJoinFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, collector: Collector[(OrderEvent, ReceiptEvent)]): Unit = {

      collector.collect((left,right))
    }
  }
}

雖然這種方法看似代碼簡單了不少，但是也存在局限性，只能匹配對應上的，不能輸出沒有匹配上的，
結果展示

小結

好了，當你看到這里的時候，意味著電商用戶行為資料分析暫時完結了，不對，下一篇文章會為大家再總結一些電商常見指標的干貨，敬請期待！！！考慮到部分小伙伴對于中間的部分代碼有疑問，所以我每行都寫上了注釋，因此詳細的程序筆者就不在這里詳細贅述了，看了注釋仍有疑惑的小伙伴們歡迎添加我的個人微信詢問，互相學習，共同進步！你知道的越多，你不知道的也越多，我是Alice，我們下一期見！

文章持續更新，可以微信搜一搜「猿人菌」第一時間閱讀，思維導圖，大資料書籍，大資料高頻面試題，海量一線大廠面經…期待您的關注!

Alice菌

CSDN認證博客專家 CSDN博客專家大資料學者追夢人

大資料學者，堅信技術成就自我，
個人原創公眾號「猿人菌」，每日更新大資料干貨，歡迎上車!
一天的生活就是一生的縮影，在最美的年華，做最好的自己！！！

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/234859.html

標籤：其他

上一篇：新書上市第13天，在亞馬遜Kindle電子書人工智能榜第三，與《未來簡史》和李開復《人工智能》同榜

下一篇：改進扇貝單詞app中的資訊可視化