0x00 文章内容

封装会话切割代码
封装会话切割逻辑

当前情况回顾，上一篇文章中我们已经实现了将输出代码重构成了一个接口组件，以达到可以选择输出TextFile格式文件或者Parquet格式文件。

现在，我们回去看一下OneUserTrackerLogsProcessor里面的代码，里面写了两个逻辑：会话切割、生成会话，而且代码也比较长，代码写在一起还不好维护，于是我们也将代码抽象出去。

0x01 封装会话切割代码

1. 抽离切割会话代码成方法

a. 当前OneUserTrackerLogsProcessor代码

package com.shaonaiyi.session
import java.net.URL
import java.util.UUID
import com.shaonaiyi.spark.session.{ TrackerLog, TrackerSession}
import org.apache.commons.lang3.time.FastDateFormat
import scala.collection.mutable.ArrayBuffer
/** * @Auther: shaonaiyi@163.com * @Date: 2019/12/14 20:38 * @Description: 转化每个user的trackerLogs为trackerSession */
class OneUserTrackerLogsProcessor(trackerLogs: Array[TrackerLog]) { 
  private val sortedTrackerLogs = trackerLogs.sortBy(trackerLog => trackerLog.getLogServerTime.toString)
  private val dateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")
  //1、会话切割
  val oneCuttingSessionLogs = new ArrayBuffer[TrackerLog]()   //存放正在切割会话的所有日志
  val initBuilder = ArrayBuffer.newBuilder[ArrayBuffer[TrackerLog]]   //存放切割完的会话的所有日志
  def buildSessions(domainLabelMap:Map[String, String]) : ArrayBuffer[TrackerSession] = { 
    val cuttedSessionLogsResult = sortedTrackerLogs.foldLeft((initBuilder, Option.empty[TrackerLog])) {  case ((builder, preLog), currLog) =>
      val currLogTime = dateFormat.parse(currLog.getLogServerTime.toString).getTime
      if (preLog.nonEmpty &&
        currLogTime - dateFormat.parse(preLog.get.getLogServerTime.toString).getTime >= 30 * 60 * 1000) { 
        //切割成新的会话
        builder += oneCuttingSessionLogs.clone()
        oneCuttingSessionLogs.clear()
      }
      oneCuttingSessionLogs += currLog
      (builder, Some(currLog))
    }._1.result()
    if (oneCuttingSessionLogs.nonEmpty) { 
      cuttedSessionLogsResult += oneCuttingSessionLogs
    }
    //2、生成会话
    cuttedSessionLogsResult.map {  case sessionLogs =>
        val session = new TrackerSession()
        session.setSessionId(UUID.randomUUID().toString)
        session.setSessionServerTime(sessionLogs.head.getLogServerTime)
        session.setCookie(sessionLogs.head.getCookie)
        session.setIp(sessionLogs.head.getIp)
        val pageviewLogs = sessionLogs.filter(_.getLogType.toString.equals("pageview"))
        if(pageviewLogs.length == 0) { 
          session.setLandingUrl("-")
        } else { 
          session.setLandingUrl(pageviewLogs.head.getUrl)
        }
        session.setPageviewCount(pageviewLogs.length)
        val clickLogs = sessionLogs.filter(_.getLogType.toString.equals("click"))
        session.setClickCount(clickLogs.length)
      if (pageviewLogs.length == 0) { 
        session.setDomain("-")
      } else { 
        val url = new URL(pageviewLogs.head.getUrl.toString)
        session.setDomain(url.getHost)
      }
      val domainLabel = domainLabelMap.getOrElse(session.getDomain.toString, "-")
      session.setDomainLabel(domainLabel)
      session
    }
  }
}

内容很多，我们需要将其进行拆分。

在拆分之前，我们先将此两行代码放进buildSessions方法里

//1、会话切割
    val oneCuttingSessionLogs = new ArrayBuffer[TrackerLog]()   //存放正在切割会话的所有日志
    val initBuilder = ArrayBuffer.newBuilder[ArrayBuffer[TrackerLog]]   //存放切割完的会话的所有日志

在这里插入图片描述

b. 抽离成切割会话方法

结果如下：
在这里插入图片描述

c. 现在虽然抽离成了一个单独的方法，但是，我们的生成规则可能是会发现变化的，到时候维护起来又特别麻烦。所以，我们可以再进一步优化，将方法抽离成Trait。

2. 抽离切割会话方法成接口

a. 新建SessionGenerator类，类型为Trait

将cutSessions方法剪切进去，因为cutSessions方法需要给其他类调用，所以此时private应该删掉，而且，此处需要用到sortedTrackerLogs，我们可以用参数传进来。
在这里插入图片描述
而且，dateFormat也要用到，也要剪切进来：

b. 修改OneUserTrackerLogsProcessor，使其继承SessionGenerator，当然，cutSessions方法也要传参数进来。如图：

3. 校验结果

a. 此时重新执行，没有报错，也能得到相应的结果。

0x02 封装会话切割逻辑

1. 抽离会话切割

场景分析：我们已经实验了按每30分钟就切割一个会话的逻辑，但实际工作中，我们的切割会话的方式可能会发现改变，比如我们可能会按pageview来切割，比如说按停留界面的时间，比如说按click事件等等。如果变一次就修改代码，这工作量就很大，所以我们跟前面一样，我们将逻辑抽离出来适配，此处以按pageview来切割为例。

a. 编写按pageview进行会话的切割，我们在SessionGenerator接口下面编写pageview逻辑，继承SessionGenerator

/** * 按照pageview进行会话的切割 */
trait PageViewSessionGenerator extends SessionGenerator { 
  override def cutSessions(sortedTrackerLogs: Array[TrackerLog]): ArrayBuffer[ArrayBuffer[TrackerLog]] = { 
    val oneCuttingSessionLogs = new ArrayBuffer[TrackerLog]() // 用于存放正在切割会话的所有的日志
    val initBuilder = ArrayBuffer.newBuilder[ArrayBuffer[TrackerLog]] // 用于存放切割完的会话所有的日志
    val cuttedSessionLogsResult: ArrayBuffer[ArrayBuffer[TrackerLog]] =
      sortedTrackerLogs.foldLeft(initBuilder) {  case (builder, currLog) =>
        // 如果当前的log是pageview的话，那么切割会话
        if (currLog.getLogType.toString.equals("pageview") && oneCuttingSessionLogs.nonEmpty) { 
          // 切割成一个新的会话
          builder += oneCuttingSessionLogs.clone()
          oneCuttingSessionLogs.clear()
        }
        // 把当前的log放到当前的会话里面
        oneCuttingSessionLogs += currLog
        builder
      }.result()
    // 最后的会话
    if (oneCuttingSessionLogs.nonEmpty) { 
      cuttedSessionLogsResult += oneCuttingSessionLogs
    }
    cuttedSessionLogsResult
  }
}

代码分析：代码与前面大同小异，需要注意的我们现在已经不需要进行日志比较时间了，所以初始参数需要删掉，并且要修改返回的结果，只返回builder。

b. 修改SessionCutETL，选择调用PageviewSessionGenerator

PS：加上with PageviewSessionGenerator表示调用

//处理每个user的日志
      val processor = new OneUserTrackerLogsProcessor(iter.toArray) with PageviewSessionGenerator

2. 校验结果

a. 为了方便观察，将保存类型修改为textfile
在这里插入图片描述
b. 执行结果

这里的结果是5个会话，因为有5个pageview日志。如果不是按pageview来切割，是3个。此处的PageviewSessionGenerator规则是直接修改代码，其实也可以写一个专门的接口来适配，通过前端所传送过来的数据，选择对应的规则切割。