一、数据说明

图片1.png
上面的数据图是从数据文件中截取的一部分内容，表示为电商网站的用户行为数据，主要包含用户的4 种行为：搜索，点击，下单，支付。数据规则如下：

  • 数据文件中每行数据采用下划线分隔数据
  • 每一行数据表示用户的一次行为，这个行为只能是 4 种行为的一种
  • 如果搜索关键字为 null,表示数据不是搜索数据
  • 如果点击的品类 ID 和产品 ID 为-1，表示数据不是点击数据
  • 针对于下单行为，一次可以下单多个商品，所以品类 ID 和产品 ID 可以是多个，id 之间采用逗号分隔，如果本次不是下单行为，则数据采用 null 表示

支付行为和下单行为类似
Ø 支付行为和下单行为类似详细字段说明：

编号 字段名称 字段类型 字段含义
1 date String 用户点击行为的日期
2 user_id Long 用户的ID
3 session_id String Session 的ID
4 page_id Long 某个页面的ID
5 action_time String 动作的时间点
6 search_keyword String 用户搜索的关键词
7 click_category_id Long 某一个商品品类的ID
8 click_product_id Long 某一个商品的ID
9 order_category_ids String 一次订单中所有品类的ID 集合
10 order_product_ids String 一次订单中所有商品的ID 集合
11 pay_category_ids String 一次支付中所有品类的ID 集合
12 pay_product_ids String 一次支付中所有商品的ID 集合
13 city_id Long 城市id

二、Top10 热门品类

本项目需求为：先按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数。

1.1、需求1-实现方案一

分别统计每个品类点击的次数，下单的次数和支付的次数：（品类，点击总数）（品类，下单总数）（品类，支付总数）

  1. package com.atguigu.bigdata.spark.core.req
  2. import org.apache.spark.rdd.RDD
  3. import org.apache.spark.{SparkConf, SparkContext}
  4. object Spark01_Req1_HotCategoryTop10Analysis {
  5.     def main(args: Array[String]): Unit = {
  6.         // TODO : Top10热门品类
  7.         val sparConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
  8.         val sc = new SparkContext(sparConf)
  10.         // 1. 读取原始日志数据
  11.         val actionRDD = sc.textFile("datas/user_visit_action.txt")
  13.         // 2. 统计品类的点击数量：（品类ID，点击数量）
  14.         val clickActionRDD = actionRDD.filter(
  15.             action => {
  16.                 val datas = action.split("_")
  17.                 datas(6) != "-1"
  18.             }
  19.         )
  21.         val clickCountRDD: RDD[(String, Int)] = clickActionRDD.map(
  22.             action => {
  23.                 val datas = action.split("_")
  24.                 (datas(6), 1)
  25.             }
  26.         ).reduceByKey(_ + _)
  28.         // 3. 统计品类的下单数量：（品类ID，下单数量）
  29.         val orderActionRDD = actionRDD.filter(
  30.             action => {
  31.                 val datas = action.split("_")
  32.                 datas(8) != "null"
  33.             }
  34.         )
  36.         // orderid => 1,2,3
  37.         // 【(1,1)，(2,1)，(3,1)】
  38.         val orderCountRDD = orderActionRDD.flatMap(
  39.             action => {
  40.                 val datas = action.split("_")
  41.                 val cid = datas(8)
  42.                 val cids = cid.split(",")
  43.                 cids.map(id=>(id, 1))
  44.             }
  45.         ).reduceByKey(_+_)
  47.         // 4. 统计品类的支付数量：（品类ID，支付数量）
  48.         val payActionRDD = actionRDD.filter(
  49.             action => {
  50.                 val datas = action.split("_")
  51.                 datas(10) != "null"
  52.             }
  53.         )
  55.         // orderid => 1,2,3
  56.         // 【(1,1)，(2,1)，(3,1)】
  57.         val payCountRDD = payActionRDD.flatMap(
  58.             action => {
  59.                 val datas = action.split("_")
  60.                 val cid = datas(10)
  61.                 val cids = cid.split(",")
  62.                 cids.map(id=>(id, 1))
  63.             }
  64.         ).reduceByKey(_+_)
  66.         // 5. 将品类进行排序，并且取前10名
  67.         //    点击数量排序，下单数量排序，支付数量排序
  68.         //    元组排序：先比较第一个，再比较第二个，再比较第三个，依此类推
  69.         //    ( 品类ID, ( 点击数量, 下单数量, 支付数量 ) )
  70.         //
  71.         //  cogroup = connect + group
  72.         val cogroupRDD: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))] = clickCountRDD.cogroup(orderCountRDD, payCountRDD)
  73.         val analysisRDD = cogroupRDD.mapValues{
  74.             case ( clickIter, orderIter, payIter ) => {
  76.                 var clickCnt = 0
  77.                 val iter1 = clickIter.iterator
  78.                 if ( iter1.hasNext ) {
  79.                     clickCnt = iter1.next()
  80.                 }
  81.                 var orderCnt = 0
  82.                 val iter2 = orderIter.iterator
  83.                 if ( iter2.hasNext ) {
  84.                     orderCnt = iter2.next()
  85.                 }
  86.                 var payCnt = 0
  87.                 val iter3 = payIter.iterator
  88.                 if ( iter3.hasNext ) {
  89.                     payCnt = iter3.next()
  90.                 }
  92.                 ( clickCnt, orderCnt, payCnt )
  93.             }
  94.         }
  95.         val resultRDD = analysisRDD.sortBy(_._2, false).take(10)
  97.         // 6. 将结果采集到控制台打印出来
  98.         resultRDD.foreach(println)
  99.         sc.stop()
  100.     }
  101. }

减少cogroup出现shuffle的影响

  1. package com.atguigu.bigdata.spark.core.req
  2. import org.apache.spark.rdd.RDD
  3. import org.apache.spark.{SparkConf, SparkContext}
  4. object Spark02_Req1_HotCategoryTop10Analysis1 {
  5.     def main(args: Array[String]): Unit = {
  6.         // TODO : Top10热门品类
  7.         val sparConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
  8.         val sc = new SparkContext(sparConf)
  10.         // Q : actionRDD重复使用
  11.         // Q : cogroup性能可能较低
  13.         // 1. 读取原始日志数据
  14.         val actionRDD = sc.textFile("datas/user_visit_action.txt")
  15.         actionRDD.cache()
  17.         // 2. 统计品类的点击数量：（品类ID，点击数量）
  18.         val clickActionRDD = actionRDD.filter(
  19.             action => {
  20.                 val datas = action.split("_")
  21.                 datas(6) != "-1"
  22.             }
  23.         )
  25.         val clickCountRDD: RDD[(String, Int)] = clickActionRDD.map(
  26.             action => {
  27.                 val datas = action.split("_")
  28.                 (datas(6), 1)
  29.             }
  30.         ).reduceByKey(_ + _)
  32.         // 3. 统计品类的下单数量：（品类ID，下单数量）
  33.         val orderActionRDD = actionRDD.filter(
  34.             action => {
  35.                 val datas = action.split("_")
  36.                 datas(8) != "null"
  37.             }
  38.         )
  40.         // orderid => 1,2,3
  41.         // 【(1,1)，(2,1)，(3,1)】
  42.         val orderCountRDD = orderActionRDD.flatMap(
  43.             action => {
  44.                 val datas = action.split("_")
  45.                 val cid = datas(8)
  46.                 val cids = cid.split(",")
  47.                 cids.map(id=>(id, 1))
  48.             }
  49.         ).reduceByKey(_+_)
  51.         // 4. 统计品类的支付数量：（品类ID，支付数量）
  52.         val payActionRDD = actionRDD.filter(
  53.             action => {
  54.                 val datas = action.split("_")
  55.                 datas(10) != "null"
  56.             }
  57.         )
  59.         // orderid => 1,2,3
  60.         // 【(1,1)，(2,1)，(3,1)】
  61.         val payCountRDD = payActionRDD.flatMap(
  62.             action => {
  63.                 val datas = action.split("_")
  64.                 val cid = datas(10)
  65.                 val cids = cid.split(",")
  66.                 cids.map(id=>(id, 1))
  67.             }
  68.         ).reduceByKey(_+_)
  70.         // (品类ID, 点击数量) => (品类ID, (点击数量, 0, 0))
  71.         // (品类ID, 下单数量) => (品类ID, (0, 下单数量, 0))
  72.         //                    => (品类ID, (点击数量, 下单数量, 0))
  73.         // (品类ID, 支付数量) => (品类ID, (0, 0, 支付数量))
  74.         //                    => (品类ID, (点击数量, 下单数量, 支付数量))
  75.         // ( 品类ID, ( 点击数量, 下单数量, 支付数量 ) )
  77.         // 5. 将品类进行排序，并且取前10名
  78.         //    点击数量排序，下单数量排序，支付数量排序
  79.         //    元组排序：先比较第一个，再比较第二个，再比较第三个，依此类推
  80.         //    ( 品类ID, ( 点击数量, 下单数量, 支付数量 ) )
  81.         //
  82.         val rdd1 = clickCountRDD.map{
  83.             case ( cid, cnt ) => {
  84.                 (cid, (cnt, 0, 0))
  85.             }
  86.         }
  87.         val rdd2 = orderCountRDD.map{
  88.             case ( cid, cnt ) => {
  89.                 (cid, (0, cnt, 0))
  90.             }
  91.         }
  92.         val rdd3 = payCountRDD.map{
  93.             case ( cid, cnt ) => {
  94.                 (cid, (0, 0, cnt))
  95.             }
  96.         }
  98.         // 将三个数据源合并在一起，统一进行聚合计算
  99.         val soruceRDD: RDD[(String, (Int, Int, Int))] = rdd1.union(rdd2).union(rdd3)
  101.         val analysisRDD = soruceRDD.reduceByKey(
  102.             ( t1, t2 ) => {
  103.                 ( t1._1+t2._1, t1._2 + t2._2, t1._3 + t2._3 )
  104.             }
  105.         )
  107.         val resultRDD = analysisRDD.sortBy(_._2, false).take(10)
  108.         // 6. 将结果采集到控制台打印出来
  109.         resultRDD.foreach(println)
  111.         sc.stop()
  112.     }
  113. }

1.2、实现方案二

一次性统计每个品类点击的次数，下单的次数和支付的次数：（品类，（点击总数，下单总数，支付总数））

  1. package com.atguigu.bigdata.spark.core.req
  2. import org.apache.spark.rdd.RDD
  3. import org.apache.spark.{SparkConf, SparkContext}
  4. object Spark03_Req1_HotCategoryTop10Analysis2 {
  5.     def main(args: Array[String]): Unit = {
  7.         // TODO : Top10热门品类
  8.         val sparConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
  9.         val sc = new SparkContext(sparConf)
  11.         // Q : 存在大量的shuffle操作（reduceByKey）
  12.         // reduceByKey 聚合算子，spark会提供优化，缓存
  14.         // 1. 读取原始日志数据
  15.         val actionRDD = sc.textFile("datas/user_visit_action.txt")
  17.         // 2. 将数据转换结构
  18.         //    点击的场合 : ( 品类ID，( 1, 0, 0 ) )
  19.         //    下单的场合 : ( 品类ID，( 0, 1, 0 ) )
  20.         //    支付的场合 : ( 品类ID，( 0, 0, 1 ) )
  21.         val flatRDD: RDD[(String, (Int, Int, Int))] = actionRDD.flatMap(
  22.             action => {
  23.                 val datas = action.split("_")
  24.                 if (datas(6) != "-1") {
  25.                     // 点击的场合
  26.                     List((datas(6), (1, 0, 0)))
  27.                 } else if (datas(8) != "null") {
  28.                     // 下单的场合
  29.                     val ids = datas(8).split(",")
  30.                     ids.map(id => (id, (0, 1, 0)))
  31.                 } else if (datas(10) != "null") {
  32.                     // 支付的场合
  33.                     val ids = datas(10).split(",")
  34.                     ids.map(id => (id, (0, 0, 1)))
  35.                 } else {
  36.                     Nil
  37.                 }
  38.             }
  39.         )
  41.         // 3. 将相同的品类ID的数据进行分组聚合
  42.         //    ( 品类ID，( 点击数量, 下单数量, 支付数量 ) )
  43.         val analysisRDD = flatRDD.reduceByKey(
  44.             (t1, t2) => {
  45.                 ( t1._1+t2._1, t1._2 + t2._2, t1._3 + t2._3 )
  46.             }
  47.         )
  49.         // 4. 将统计结果根据数量进行降序处理，取前10名
  50.         val resultRDD = analysisRDD.sortBy(_._2, false).take(10)
  51.         // 5. 将结果采集到控制台打印出来
  52.         resultRDD.foreach(println)
  53.         sc.stop()
  54.     }
  55. }

1.3、实现方案三

使用累加器的方式聚合数据

  1. package com.atguigu.bigdata.spark.core.req
  2. import org.apache.spark.rdd.RDD
  3. import org.apache.spark.util.AccumulatorV2
  4. import org.apache.spark.{SparkConf, SparkContext}
  5. import scala.collection.mutable
  6. object Spark04_Req1_HotCategoryTop10Analysis3 {
  7.     def main(args: Array[String]): Unit = {
  8.         // TODO : Top10热门品类
  9.         val sparConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
  10.         val sc = new SparkContext(sparConf)
  12.         // 1. 读取原始日志数据
  13.         val actionRDD = sc.textFile("datas/user_visit_action.txt")
  15.         val acc = new HotCategoryAccumulator
  16.         sc.register(acc, "hotCategory")
  18.         // 2. 将数据转换结构
  19.         actionRDD.foreach(
  20.             action => {
  21.                 val datas = action.split("_")
  22.                 if (datas(6) != "-1") {
  23.                     // 点击的场合
  24.                     acc.add((datas(6), "click"))
  25.                 } else if (datas(8) != "null") {
  26.                     // 下单的场合
  27.                     val ids = datas(8).split(",")
  28.                     ids.foreach(
  29.                         id => {
  30.                             acc.add( (id, "order") )
  31.                         }
  32.                     )
  33.                 } else if (datas(10) != "null") {
  34.                     // 支付的场合
  35.                     val ids = datas(10).split(",")
  36.                     ids.foreach(
  37.                         id => {
  38.                             acc.add( (id, "pay") )
  39.                         }
  40.                     )
  41.                 }
  42.             }
  43.         )
  45.         val accVal: mutable.Map[String, HotCategory] = acc.value
  46.         val categories: mutable.Iterable[HotCategory] = accVal.map(_._2)
  48.         val sort = categories.toList.sortWith(
  49.             (left, right) => {
  50.                 if ( left.clickCnt > right.clickCnt ) {
  51.                     true
  52.                 } else if (left.clickCnt == right.clickCnt) {
  53.                     if ( left.orderCnt > right.orderCnt ) {
  54.                         true
  55.                     } else if (left.orderCnt == right.orderCnt) {
  56.                         left.payCnt > right.payCnt
  57.                     } else {
  58.                         false
  59.                     }
  60.                 } else {
  61.                     false
  62.                 }
  63.             }
  64.         )
  66.         // 5. 将结果采集到控制台打印出来
  67.         sort.take(10).foreach(println)
  69.         sc.stop()
  70.     }
  72.     case class HotCategory( cid:String, var clickCnt : Int, var orderCnt : Int, var payCnt : Int )
  74.       /**
  75.       * 自定义累加器
  76.       * 1. 继承AccumulatorV2，定义泛型
  77.       *    IN : ( 品类ID, 行为类型 )
  78.       *    OUT : mutable.Map[String, HotCategory]
  79.       * 2. 重写方法（6）
  80.       */
  81.     class HotCategoryAccumulator extends AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]{
  83.         private val hcMap = mutable.Map[String, HotCategory]()
  85.         override def isZero: Boolean = {
  86.             hcMap.isEmpty
  87.         }
  89.         override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] = {
  90.             new HotCategoryAccumulator()
  91.         }
  93.         override def reset(): Unit = {
  94.             hcMap.clear()
  95.         }
  97.         override def add(v: (String, String)): Unit = {
  98.             val cid = v._1
  99.             val actionType = v._2
  100.             val category: HotCategory = hcMap.getOrElse(cid, HotCategory(cid, 0,0,0))
  101.             if ( actionType == "click" ) {
  102.                 category.clickCnt += 1
  103.             } else if (actionType == "order") {
  104.                 category.orderCnt += 1
  105.             } else if (actionType == "pay") {
  106.                 category.payCnt += 1
  107.             }
  108.             hcMap.update(cid, category)
  109.         }
  111.         override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]): Unit = {
  112.             val map1 = this.hcMap
  113.             val map2 = other.value
  115.             map2.foreach{
  116.                 case ( cid, hc ) => {
  117.                     val category: HotCategory = map1.getOrElse(cid, HotCategory(cid, 0,0,0))
  118.                     category.clickCnt += hc.clickCnt
  119.                     category.orderCnt += hc.orderCnt
  120.                     category.payCnt += hc.payCnt
  121.                     map1.update(cid, category)
  122.                 }
  123.             }
  124.         }
  126.         override def value: mutable.Map[String, HotCategory] = hcMap
  127.     }
  128. }

三、需求2-Top10 热门品类中每个品类的Top10 活跃Session 统计

在需求一的基础上，增加每个品类用户session 的点击统计

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark05_Req2_HotCategoryTop10SessionAnalysis {

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

        // TODO : Top10热门品类
        val sparConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
        val sc = new SparkContext(sparConf)

        val actionRDD = sc.textFile("datas/user_visit_action.txt")
        actionRDD.cache()
        val top10Ids: Array[String] = top10Category(actionRDD)

        // 1. 过滤原始数据,保留点击和前10品类ID
        val filterActionRDD = actionRDD.filter(
            action => {
                val datas = action.split("_")
                if ( datas(6) != "-1" ) {
                    top10Ids.contains(datas(6))
                } else {
                    false
                }
            }
        )

        // 2. 根据品类ID和sessionid进行点击量的统计
        val reduceRDD: RDD[((String, String), Int)] = filterActionRDD.map(
            action => {
                val datas = action.split("_")
                ((datas(6), datas(2)), 1)
            }
        ).reduceByKey(_ + _)

        // 3. 将统计的结果进行结构的转换
        //  (（ 品类ID，sessionId ）,sum) => ( 品类ID，（sessionId, sum） )
        val mapRDD = reduceRDD.map{
            case ( (cid, sid), sum ) => {
                ( cid, (sid, sum) )
            }
        }

        // 4. 相同的品类进行分组
        val groupRDD: RDD[(String, Iterable[(String, Int)])] = mapRDD.groupByKey()

        // 5. 将分组后的数据进行点击量的排序，取前10名
        val resultRDD = groupRDD.mapValues(
            iter => {
                iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
            }
        )

        resultRDD.collect().foreach(println)
        sc.stop()
    }

    def top10Category(actionRDD:RDD[String]) = {
        val flatRDD: RDD[(String, (Int, Int, Int))] = actionRDD.flatMap(
            action => {
                val datas = action.split("_")
                if (datas(6) != "-1") {
                    // 点击的场合
                    List((datas(6), (1, 0, 0)))
                } else if (datas(8) != "null") {
                    // 下单的场合
                    val ids = datas(8).split(",")
                    ids.map(id => (id, (0, 1, 0)))
                } else if (datas(10) != "null") {
                    // 支付的场合
                    val ids = datas(10).split(",")
                    ids.map(id => (id, (0, 0, 1)))
                } else {
                    Nil
                }
            }
        )

        val analysisRDD = flatRDD.reduceByKey(
            (t1, t2) => {
                ( t1._1+t2._1, t1._2 + t2._2, t1._3 + t2._3 )
            }
        )
        analysisRDD.sortBy(_._2, false).take(10).map(_._1)
    }
}

四、需求3-页面单跳转换率统计

4.1、页面单跳转化率

计算页面单跳转化率，什么是页面单跳转换率，比如一个用户在一次Session 过程中访问的页面路径3,5,7,9,10,21，那么页面3 跳到页面5 叫一次单跳，7-9 也叫一次单跳， 那么单跳转化率就是要统计页面点击的概率。
比如：计算3-5 的单跳转化率，先获取符合条件的Session 对于页面3 的访问次数（PV）为 A，然后获取符合条件的 Session 中访问了页面 3 又紧接着访问了页面 5 的次数为 B，那么B/A 就是3-5 的页面单跳转化率。

4.2、统计页面单跳转化率意义

产品经理和运营总监，可以根据这个指标，去尝试分析，整个网站，产品，各个页面的表现怎么样，是不是需要去优化产品的布局；吸引用户最终可以进入最后的支付页面。
数据分析师，可以此数据做更深一步的计算和分析。
企业管理层，可以看到整个公司的网站，各个页面的之间的跳转的表现如何，可以适当调整公司的经营战略或策略。
图片1.png

4.3、代码实现

package com.atguigu.bigdata.spark.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark06_Req3_PageflowAnalysis {

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

        // TODO : Top10热门品类
        val sparConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
        val sc = new SparkContext(sparConf)

        val actionRDD = sc.textFile("datas/user_visit_action.txt")

        val actionDataRDD = actionRDD.map(
            action => {
                val datas = action.split("_")
                UserVisitAction(
                    datas(0),
                    datas(1).toLong,
                    datas(2),
                    datas(3).toLong,
                    datas(4),
                    datas(5),
                    datas(6).toLong,
                    datas(7).toLong,
                    datas(8),
                    datas(9),
                    datas(10),
                    datas(11),
                    datas(12).toLong
                )
            }
        )
        actionDataRDD.cache()

        // TODO 对指定的页面连续跳转进行统计
        // 1-2,2-3,3-4,4-5,5-6,6-7
        val ids = List[Long](1,2,3,4,5,6,7)
        val okflowIds: List[(Long, Long)] = ids.zip(ids.tail)

        // TODO 计算分母
        val pageidToCountMap: Map[Long, Long] = actionDataRDD.filter(
            action => {
                ids.init.contains(action.page_id)
            }
        ).map(
            action => {
                (action.page_id, 1L)
            }
        ).reduceByKey(_ + _).collect().toMap

        // TODO 计算分子

        // 根据session进行分组
        val sessionRDD: RDD[(String, Iterable[UserVisitAction])] = actionDataRDD.groupBy(_.session_id)

        // 分组后，根据访问时间进行排序（升序）
        val mvRDD: RDD[(String, List[((Long, Long), Int)])] = sessionRDD.mapValues(
            iter => {
                val sortList: List[UserVisitAction] = iter.toList.sortBy(_.action_time)

                // 【1，2，3，4】
                // 【1，2】，【2，3】，【3，4】
                // 【1-2，2-3，3-4】
                // Sliding : 滑窗
                // 【1，2，3，4】
                // 【2，3，4】
                // zip : 拉链
                val flowIds: List[Long] = sortList.map(_.page_id)
                val pageflowIds: List[(Long, Long)] = flowIds.zip(flowIds.tail)

                // 将不合法的页面跳转进行过滤
                pageflowIds.filter(
                    t => {
                        okflowIds.contains(t)
                    }
                ).map(
                    t => {
                        (t, 1)
                    }
                )
            }
        )
        // ((1,2),1)
        val flatRDD: RDD[((Long, Long), Int)] = mvRDD.map(_._2).flatMap(list=>list)
        // ((1,2),1) => ((1,2),sum)
        val dataRDD = flatRDD.reduceByKey(_+_)

        // TODO 计算单跳转换率
        // 分子除以分母
        dataRDD.foreach{
            case ( (pageid1, pageid2), sum ) => {
                val lon: Long = pageidToCountMap.getOrElse(pageid1, 0L)

                println(s"页面${pageid1}跳转到页面${pageid2}单跳转换率为:" + ( sum.toDouble/lon ))
            }
        }
        sc.stop()
    }

    //用户访问动作表
    case class UserVisitAction(
              date: String,//用户点击行为的日期
              user_id: Long,//用户的ID
              session_id: String,//Session的ID
              page_id: Long,//某个页面的ID
              action_time: String,//动作的时间点
              search_keyword: String,//用户搜索的关键词
              click_category_id: Long,//某一个商品品类的ID
              click_product_id: Long,//某一个商品的ID
              order_category_ids: String,//一次订单中所有品类的ID集合
              order_product_ids: String,//一次订单中所有商品的ID集合
              pay_category_ids: String,//一次支付中所有品类的ID集合
              pay_product_ids: String,//一次支付中所有商品的ID集合
              city_id: Long
      )//城市 id
}