基于spark的用户行为路径分析

import java.util

import com.google.gson.Gson
import org.apache.spark.SparkContext
import org.apache.log4j.{Level, Logger => LG}
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.hive.HiveContext

/**
  * 用户行为路径实时计算实现
  * Created by chouyarn on 2016/12/12.
  */

/**
  * 树结构类
  *
  * @param name      页面路径
  * @param visit     访次
  * @param pv        pv
  * @param uv        uv
  * @param childList 儿子链表
  */
class Node(
            var name: String,
            var path:Any,
            var visit: Any,
            var pv: Any,
            var uv: Any,
            var childList: util.ArrayList[Node]) extends Serializable {
  /**
    * 添加子节点
    *
    * @param node 子节点对象
    * @return
    */
  def addNode(node: Node) = {
    childList.add(node)
  }

  /**
    * 遍历节点，深度优先
    */
  def traverse(): Unit = {
    if (childList.isEmpty)
      return
    //    node.
    val childNum = childList.size

    for (i <- 0 to childNum - 1) {
      val child: Node = childList.get(i)
      child.name = child.name.split("_").last//去除前边绝对路径
      child.traverse()
    }
  }

  /**
    * 遍历节点，深度优先
    */
  def traverse(pages:DataFrame): Unit = {
    if (childList.isEmpty||childList.size()==0)
      return
    //    node.
    val childNum = childList.size

    for (i <- 0 to childNum - 1) {
      val child: Node = childList.get(i)
      child.name = child.name.split("_").last
      val id =pages.filter("page_id='"+child.name+"'").select("page_name").first().getString(0)//替换id为name
      child.name = id
      child.traverse(pages)
    }
  }

  /**
    * 动态插入节点
    *
    * @param node 节点对象
    * @return
    */
  def insertNode(node: Node): Boolean = {
    val insertName = node.name
    if (insertName.stripSuffix("_" + insertName.split("_").last).equals(name)) {
      //      node.name=node.name.split("_").last
      addNode(node)
      true
    } else {
      val childList1 = childList
      val childNum = childList1.size
      var insetFlag = false
      for (i <- 0 to childNum - 1) {
        val childNode = childList1.get(i)
        insetFlag = childNode.insertNode(node)
        if (insetFlag == true)
          true
      }
      false
    }
  }
}

/**
  * 处理类
  */
class Path extends CleanDataWithRDD {
  LG.getRootLogger.setLevel(Level.ERROR)//控制spark日志输出级别

  val sc: SparkContext = SparkUtil.createSparkContextYarn("path")
  val hiveContext = new HiveContext(sc)

  override def handleData(conf: Map[String, String]): Unit = {

    val num = conf.getOrElse("depth", 5)//路径深度
    val pageName = conf.getOrElse("pageName", "")//页面名称
    //    val pageName = "A_C"
    val src = conf.getOrElse("src", "")//标示来源pc or wap

    val pageType = conf.getOrElse("pageType", "")//向前或者向后路径
    val startDate = conf.getOrElse("startDate", "")//开始日期
    val endDate = conf.getOrElse("endDate", "")//结束日期
    //        保存log缓存以保证后续使用
    val log = hiveContext.sql(s"select fpid,sessionid,path " +
      s"from specter.t_pagename_path_sparksource " +
      s"where day between '$startDate' and '$endDate' and path_type=$pageType and src='$src' ")
      .map(s => {
        (s.apply(0) + "_" + s.apply(1) + "_" + s.apply(2))
      }).repartition(10).persist()

    val pages=hiveContext.sql("select page_id,page_name from specter.code_pagename").persist()//缓存页面字典表
    // 本地测试数据
    // val log = sc.parallelize(Seq("fpid1_sessionid1_A_B",
    //      "fpid2_sessionid2_A_C_D_D_B_A_D_A_F_B",
    //      "fpid1_sessionid1_A_F_A_C_D_A_B_A_V_A_N"))
    var root: Node = null
    /**
      * 递归将计算的节点放入树结构
      *
      * @param pageName 页面名称
      */
    def compute(pageName: String): Unit = {
      val currenRegex = pageName.r //页面的正则表达式
      val containsRdd = log.filter(_.contains(pageName)).persist() //包含页面名称的RDD，后续步骤用到
      val currentpv = containsRdd.map(s => {//计算pv
        currenRegex findAllIn (s)
      }).map(_.mkString(","))
        .flatMap(_.toString.split(","))
        .filter(_.size > 0)
        .count()

      val tempRdd = containsRdd.map(_.split("_")).persist() //分解后的RDD
      val currentuv = tempRdd.map(_.apply(0)).distinct().count() //页面uv
      val currentvisit = tempRdd.map(_.apply(1)).distinct().count() //页面访次

      //      初始化根节点或添加节点
      if (root == null) {
        root = new Node(pageName,pageName.hashCode, currentvisit, currentpv, currentuv, new util.ArrayList[Node]())
      } else {
        root.insertNode(new Node(pageName,pageName.hashCode, currentvisit, currentpv, currentuv, new util.ArrayList[Node]()))
      }

      if (pageName.split("_").size == 5||tempRdd.isEmpty()) {//递归出口
        return
      } else {
        //          确定下个页面名称正则表达式
        val nextRegex =
        s"""${pageName}_[0-9a-z]{8}-[0-9a-z]{4}-[0-9a-z]{4}-[0-9a-z]{4}-[0-9a-z]{12}""".r
        // 本地测试       
        // val nextRegex =s"""${pageName}_[A-Z]""".r
        val nextpvMap = containsRdd.map(s => {//下一级路径的pv数top9
          nextRegex findAllIn (s)
        }).map(_.mkString(","))
          .flatMap(_.toString.split(","))
          .filter(_.size > 0)
          .map(s => (s.split("_").last, 1))
          .filter(!_._1.contains(pageName.split("_")(0)))
          .reduceByKey(_ + _).sortBy(_._2, false).take(9).toMap

        nextpvMap.keySet.foreach(key => {//递归计算
          compute(pageName + "_" + key)
        })
      }
    }
    //触发计算
    compute(pageName)
    val gson: Gson = new Gson()

    root.traverse(pages)
    root.name=pages.filter("page_id='"+pageName+"'").select("page_name").first().getString(0)
    println(gson.toJson(root))//转化成JSON并打印，Alibaba fsatjson不可用，还是google得厉害。
  }

  override def stop(): Unit = {
    sc.stop()
  }
}

object Path {
  def main(args: Array[String]): Unit = {
    //    println("ss".hashCode)
    var num=5
    try {
      num=args(5).toInt
    }catch {
      case e:Exception =>
    }

    val map = Map("pageName" -> args(0),
      "pageType" -> args(1),
      "startDate" -> args(2),
      "endDate" -> args(3),
      "src" -> args(4),
      "depth" -> num.toString)
    val path = new Path()
    path.handleData(map)
  }
}