spark项目实战-电商分析平台

第1章 项目概述

电商分析平台是对用户访问电商平台的行为进行分析。

1.1 项目简介

  本项目主要讲解一个大型电商网站后台的企业级大数据统计分析平台，该平台以 Spark 为主，对电商网站的流量进行离线和实时的分析。
  该大数据分析平台对电商网站的各类用户行为（访问行为、购物行为、广告点击行为等）进行复杂的分析。用统计分析出来的数据，辅助公司中的 PM（产品经理）、数据分析师以及管理人员分析现有产品的状况，并根据用户行为分析结果持续改进产品的设计，以及调整公司的战略和业务。最终达到用大数据技术来帮助提高公司的业绩、营业额以及市场占有率的目标。
  项目主要使用了 Spark 技术生态栈中最经常使用的三个技术框架，Spark Core、Spark SQL 和 Spark Streaming，进行离线计算和实时计算业务模块的开发。实现了包括用户访问 session 分析、页面单跳转化率统计、热门商品离线统计、广告流量实时统计 4 个业务模块。
  项目中全部的业务功能模块都是直接从实际企业项目中抽取出来的，业务复杂度没有任何缩水，经过合理的将实际业务模块进行技术整合与改造，该项目几乎彻底涵盖了 Spark Core、Spark SQL 和 Spark Streaming 这三个技术框架中大部分的功能点、知识点。mysql

1.2 项目目标

  一、掌握电商系统中 Spark 的主要使用场景以及建设流程。
  二、掌握企业级的 Spark 项目的复杂性能调优、线上故障解决经验、数据倾斜全套处理方案。
  三、经过项目实战，彻底将 Spark 全部技术点和知识点都应用在项目中，掌握如何灵活应用 Spark 各项技术来实现各类复杂业务需求。nginx

1.3 业务需求简介

1.3.1 用户访问 session 统计

  用户在电商网站上，一般会有不少的访问行为，一般都是进入首页，而后可能点击首页上的一些商品，点击首页上的一些品类，也可能随时在搜索框里面搜索关键词，还可能将一些商品加入购物车，对购物车中的多个商品下订单，最后对订单中的多个商品进行支付。
  用户的每一次操做，其实能够理解为一个 action，在本项目中，咱们关注点击、搜索、下单、支付这四个用户行为。
  用户 session，是在电商平台的角度定义的会话概念，指的就是，从用户第一次进入首页，session 就开始了。而后在必定时间范围内，直到最后操做完（可能作了几十次、甚至上百次操做），离开网站，关闭浏览器，或者长时间没有作操做，那么 session 就结束了。
  以上用户在网站内的访问过程，就称之为一次 session。简单理解，session 就是某一天某一个时间段内，某个用户对网站从打开/进入，到作了大量操做，到最后关闭浏览器的过程，就叫作 session。
  session 实际上就是一个电商网站中最基本的数据。那么面向消费者/用户端的大数据分析（C端），最基本的就是面向用户访问行为/用户访问 session 的分析。
  该模块主要是对用户访问 session 进行统计分析，包括 session 的聚合指标计算、按时间比例随机抽取 session、获取天天点击、下单和购买排名前 10 的品类、并获取 top10 品类的点击量排名前 10 的 session。该模块可让产品经理、数据分析师以及企业管理层形象地看到各类条件下的具体用户行为以及统计指标，从而对公司的产品设计以及业务发展战略作出调整。主要使用 Spark Core 实现。web

1.3.2 页面单跳转化率统计

  页面单跳转化率是一个很是有用的统计数据。
  产品经理，能够根据这个指标，去尝试分析整个网站/产品，各个页面的表现怎么样，是否是须要去优化产品的布局；吸引用户最终能够进入最后的支付页面。
  数据分析师，能够基于此数据，作更深一步的计算和分析。
  企业管理层，能够看到整个公司的网站，各个页面的之间的跳转的表现如何，作到内心有数，能够适当调整公司的经营战略或策略。
  该模块主要是计算关键页面之间的单步跳转转化率，涉及到页面切片算法以及页面流匹配算法。该模块可让产品经理、数据分析师以及企业管理层看到各个关键页面之间的转化率，从而对网页布局，进行更好的优化设计。主要使用 Spark Core 实现。算法

1.3.3 区域热门商品离线统计

  该模块主要实现天天统计出各个区域的 top3 热门商品。
  咱们认为，不一样地区的经济发展水平不一样，地理环境及气候不一样，人们的风土人情和风俗习惯不一样，所以对于不一样商品的需求不一样，根据区域热门商品的统计，可让公司决策层更好的对不一样类型商品进行布局，使商品进入最须要他的区域。
  该模块可让企业管理层看到公司售卖的商品的总体状况，从而对公司的商品相关的战略进行调整。主要使用 Spark SQL 实现。sql

1.3.4 广告流量实时统计

  网站/app 中常常会给第三方平台作广告，这也是一些互联网公司的核心收入来源；当广告位招商完成后，广告会在 网站/app 的某个广告位发布出去，当用户访问 网站/app 的时候，会看到相应位置的广告，此时，有些用户可能就会去点击那个广告。
  咱们要获取用户点击广告的行为，并针对这一行为进行计算和统计。
  用户每次点击一个广告之后，会产生相应的埋点日志；在大数据实时统计系统中，会经过某些方式将数据写入到分布式消息队列中（Kafka）。
  日志发送给后台 web 服务器（nginx），nginx 将日志数据负载均衡到多个 Tomcat 服务器上，Tomcat 服务器会不断将日志数据写入 Tomcat 日志文件中，写入后，就会被日志采集客户端（好比 Flume Agent）所采集，随后写入到消息队列中（Kafka），咱们的实时计算程序会从消息队列中（ Kafka）去实时地拉取数据，而后对数据进行实时的计算和统计。
Kafka这个模块的意义在于，让产品经理、高管能够实时地掌握到公司打的各类广告的投放效果。以便于后期持续地对公司的广告投放相关的战略和策略，进行调整和优化；以指望得到最好的广告收益。
  该模块负责实时统计公司的广告流量，包括广告展示流量和广告点击流量。实现动态黑名单机制，以及黑名单过滤；实现滑动窗口内的各城市的广告展示流量和广告点击流量的统计；实现每一个区域每一个广告的点击流量实时统计；实现每一个区域 top3 点击量的广告的统计。主要使用 Spark Streaming 实现。数据库

第2章 项目主体架构

2.1 项目架构

  用户行为数据在网站上最简单的存在形式就是日志。网站在运行过程当中会产生大量的原始日志 RAW LOG，将其存储在文件系统中，企业会将多种原始日志按照用户行为汇总成会话日志 SESSION LOG，每个会话日志表示用户的一种反馈。
  本项目分为离线分析系统与实时分析系统两大模块。
  在离线分析系统中，咱们将模拟业务数据写入 Hive 表中，离线分析系统从 Hive 中获取数据，并根据实际需求（用户访问 Session 分析、页面单跳转化率分析、各区域热门商品统计） 对数据进行处理，最终将分析完毕的统计数据存储到 MySQL 的对应表格中。
  在实时分析系统中，咱们将模拟业务数据写入 Kafka 集群中， 实时分析系统从 Kafka broker 中获取数据，经过 Spark Streaming 的流式处理对广告点击流量进行实时分析，最终将统计结果存储到 MySQL 的对应表格中。express

2.2 离线日志采集宏观流程（参考）

2.3 实时日志采集宏观流程（参考）

2.4 离线/实时日志采集框架

  上图是一个企业级的日志处理框架，这一框架实现了对日志信息进行采集、汇总、清洗、聚合、分析的完整过程，并将日志数据分别存储到了离线和实时数据处理模块中，使得分析系统能够经过离线和实时两个角度对数据进行分析统计，并根据统计结果指导业务平台的改良和优化。apache

第3章 模拟业务数据源

3.1 离线数据

举例

3.1.1 数据模型与数听说明

一、user_visit_action
user_visit_action 表，存放网站或者 APP 天天的点击流数据。通俗地讲，就是用户对 网站/APP 每点击一下，就会产生一条存放在这个表里面的数据。

user_visit_action 表中的字段解析以下所示：

字段名称            说明
date                日期，表明这个用户点击行为是在哪一天发生的
user_id             用户 ID，惟一地标识某个用户
session_id          Session ID，惟一地标识某个用户的一个访问 session
page_id             页面 ID，点击了某些商品/品类，也多是搜索了某个关键词，而后进入了某个页面，页面的 id
action_time         动做时间，这个点击行为发生的时间点
search_keyword      搜索关键词，若是用户执行的是一个搜索行为，好比说在 网站/app 中，搜索了某个关键词，而后会跳转到商品列表页面
click_category_id   点击品类 ID，多是在网站首页，点击了某个品类（美食、电子设备、电脑）
click_product_id    点击商品 ID，多是在网站首页，或者是在商品列表页，点击了某个商品（好比呷哺呷哺火锅 XX 路店 3 人套餐、iphone 6s）
order_category_ids  下单品类 ID，表明了可能将某些商品加入了购物车，而后一次性对购物车中的商品下了一个订单，这就表明了某次下单的行为中，有哪些商品品类，可能有 6 个商品，可是就对应了 2 个品类，好比有 3 根火腿肠（食品品类），3 个电池（日用品品类）
order_product_ids   下单商品 ID，某次下单，具体对哪些商品下的订单
pay_category_ids    付款品类 ID，对某个订单，或者某几个订单，进行了一次支付的行为，对应了哪些品类
pay_product_ids     付款商品 ID，支付行为下，对应的哪些具体的商品
city_id             城市 ID，表明该用户行为发生在哪一个城市

二、user_info
user_info 表，是一张普通的用户基本信息表；这张表中存放了 网站/APP 全部注册用户的基本信息。

user_info 表中的字段解析以下所示：

字段名称            说明
user_id             用户 ID，惟一地标识某个用户
username            用户登陆名
name                用户昵称或真实姓名
age                 用户年龄
professional        用户职业
city                用户所在城市
sex                 用户性别

三、product_info
product_info 表，是一张普通的商品基本信息表；这张表中存放了 网站/APP 全部商品的基本信息。

product_info 表中的字段解析以下所示：

字段名称            说明
proudct_id          商品 ID，惟一地标识某个商品
product_name        商品名称
extend_info         额外信息，例如商品为自营商品仍是第三方商品

3.2 实时数据

3.2.1 数据模型与数听说明

程序每 5 秒向 Kafka 集群写入数据，格式以下：
格式 ：timestamp province city userid adid
在线数据的字段解析以下所示：

字段名称            取值范围
timestamp           当前时间毫秒
userId              0 – 99
provice/city        1 – 9((0L," 北京"," 北京"),(1L," 上海"," 上海"),(2L," 南京"," 江苏省"),(3L,"广州","广东省"),(4L,"三亚","海南省"),(5L,"武汉","湖北省"),(6L,"长沙","湖南省"),(7L,"西安","陕西省"),(8L,"成都","四川省"),(9L,"哈尔滨","东北省"))
adid                0 - 19

第4章 程序框架解析

新建一个 maven 工程 commerce_basic 做为父 maven 工程，引入依赖 pom.xml
pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
 
    <groupId>com.atguigu</groupId>
    <artifactId>commerce</artifactId>
    <packaging>pom</packaging>
    <version>1.0-SNAPSHOT</version>
    <modules>
        <module>commons</module>
        <module>mock</module>
    </modules>
 
    <!-- 声明子项目公用的配置属性 -->
    <properties>
        <spark.version>2.1.1</spark.version>
        <scala.version>2.11.8</scala.version>
        <log4j.version>1.2.17</log4j.version>
        <slf4j.version>1.7.22</slf4j.version>
    </properties>
 
    <!-- 声明并引入子项目共有的依赖 -->
    <dependencies>
        <!-- 全部子项目的日志框架 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>jcl-over-slf4j</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <!-- 具体的日志实现 -->
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        <!-- Logging End -->
    </dependencies>
    <dependencyManagement>
        <dependencies>
            <!-- 引入 Spark 相关的 Jar 包 -->
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.11</artifactId>
                <version>${spark.version}</version>
                <!-- provider 若是存在，那么运行时该 Jar 包不存在，也不会打包到最终的发布版本中，只是编译器有效 -->
                <!--<scope>provided</scope>-->
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-sql_2.11</artifactId>
                <version>${spark.version}</version>
                <!--<scope>provided</scope>-->
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming_2.11</artifactId>
                <version>${spark.version}</version>
                <!--<scope>provided</scope>-->
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-mllib_2.11</artifactId>
                <version>${spark.version}</version>
                <!--<scope>provided</scope>-->
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-graphx_2.11</artifactId>
                <version>${spark.version}</version>
                <!--<scope>provided</scope>-->
            </dependency>
            <dependency>
                <groupId>org.scala-lang</groupId>
                <artifactId>scala-library</artifactId>
                <version>${scala.version}</version>
                <!--<scope>provided</scope>-->
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-hive_2.11</artifactId>
                <version>${spark.version}</version>
                <!--<scope>provided</scope>-->
            </dependency>
        </dependencies>
    </dependencyManagement>
 
    <!-- 声明构建信息 -->
    <build>
        <!-- 声明并引入子项目共有的插件：插件就是负责到 Maven 各个声明周期的具体实现 -->
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.1</version>
                <!-- 全部的编译都依照 JDK1.8 来搞 -->
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
 
        <!-- 仅声明子项目共有的插件，若是子项目须要此插件，那么子项目须要声明 -->
        <pluginManagement>
            <plugins>
                <!-- 该插件用于将 Scala 代码编译成 class 文件 -->
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.2.2</version>
                    <executions>
                        <execution>
                            <!-- 声明绑定到 maven 的 compile 阶段 -->
                            <goals>
                                <goal>compile</goal>
                                <goal>testCompile</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
                <!-- 用于项目的打包插件 -->
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-assembly-plugin</artifactId>
                    <version>3.0.0</version>
                    <executions>
                        <execution>
                            <id>make-assembly</id>
                            <phase>package</phase>
                            <goals>
                                <goal>single</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </pluginManagement>
    </build>
</project>

4.1 mock 模块(模拟数据产生模块)

新建一个模块 maven 工程 mock 做为子 maven 工程，引入依赖 pom.xml
pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
 
    <parent>
        <artifactId>commerce</artifactId>
        <groupId>com.atguigu</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
 
    <artifactId>mock</artifactId>
 
    <dependencies>
        <dependency>
            <groupId>com.atguigu</groupId>
            <artifactId>commons</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <!-- Spark 的依赖引入 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.2.1</version>
        </dependency>
    </dependencies>
 
    <build>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

类(包)名称

类(包)结构图

MockDataGenerate.scala

import java.util.UUID
 
import commons.model.{ProductInfo, UserInfo, UserVisitAction}
import commons.utils.{DateUtils, StringUtils}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
 
import scala.collection.mutable.ArrayBuffer
import scala.util.Random
 
/**
  * 离线模拟数据的生成
  *
  * date：是当前日期
  * age: 0 - 59
  * professionals: professional[0 - 99]
  * cities: 0 - 99
  * sex: 0 - 1
  * keywords: ("火锅", "蛋糕", "重庆辣子鸡", "重庆小面", "呷哺呷哺", "新辣道鱼火锅", "国贸大厦", "太古商场", "日本料理", "温泉")
  * categoryIds: 0 - 99
  * ProductId: 0 - 99
  */
object MockDataGenerate {
  /**
    * 模拟用户的行为信息
    *
    * @return
    */
  private def mockUserVisitActionData(): Array[UserVisitAction] = {
    val rows = ArrayBuffer[UserVisitAction]()
    val random = new Random()
 
    val searchKeywords = Array("华为手机", "联想笔记本", "小龙虾", "卫生纸", "吸尘器", "Lamer", "机器学习", "苹果", "洗面奶", "保温杯")
    // yyyy-MM-dd
    val date = DateUtils.getTodayDate()
    // 关注四个行为：搜索、点击、下单、支付
    val actions = Array("search", "click", "order", "pay")
 
    // 一共 100 个用户（有重复）
    for (i <- 0 until 100) {
      val userid = random.nextInt(100)
      // 每一个用户产生 10 个 session
      for (j <- 0 until 10) {
        // 不可变的，全局的，独一无二的 128bit 长度的标识符，用于标识一个 session，体现一次会话产生的 sessionId 是独一无二的
        val sessionid = UUID.randomUUID().toString().replace("-", "")
        // 在 yyyy-MM-dd 后面添加一个随机的小时时间（0-23）
        val baseActionTime = date + " " + random.nextInt(23) // 2019-05-30 12
        // 每一个 (userid + sessionid) 生成 0-100 条用户访问数据
        for (k <- 0 to random.nextInt(100)) {
          val pageid = random.nextInt(10)
          // 在 yyyy-MM-dd HH 后面添加一个随机的分钟时间和秒时间，2019-05-30 12:25:30
          val actionTime = baseActionTime + ":" + StringUtils.fulfuill(String.valueOf(random.nextInt(59))) + ":" + StringUtils.fulfuill(String.valueOf(random.nextInt(59)))
          var searchKeyword: String = null
          var clickCategoryId: Long = -1L
          var clickProductId: Long = -1L
          var orderCategoryIds: String = null
          var orderProductIds: String = null
          var payCategoryIds: String = null
          var payProductIds: String = null
          val cityid = random.nextInt(10).toLong
 
          // 随机肯定用户在当前 session 中的行为
          val action = actions(random.nextInt(4))
 
          // 根据随机产生的用户行为 action 决定对应字段的值
          action match {
            case "search" => searchKeyword = searchKeywords(random.nextInt(10))
            case "click" => clickCategoryId = random.nextInt(100).toLong
              clickProductId = String.valueOf(random.nextInt(100)).toLong
            case "order" => orderCategoryIds = random.nextInt(100).toString
              orderProductIds = random.nextInt(100).toString
            case "pay" => payCategoryIds = random.nextInt(100).toString
              payProductIds = random.nextInt(100).toString
          }
 
          rows += UserVisitAction(date, userid, sessionid,
            pageid, actionTime, searchKeyword,
            clickCategoryId, clickProductId,
            orderCategoryIds, orderProductIds,
            payCategoryIds, payProductIds, cityid)
        }
      }
    }
 
    rows.toArray
  }
 
  /**
    * 模拟用户信息表
    *
    * @return
    */
  private def mockUserInfo(): Array[UserInfo] = {
    val rows = ArrayBuffer[UserInfo]()
 
    val sexes = Array("male", "female")
    val random = new Random()
 
    // 随机产生 100 个用户的我的信息
    for (i <- 0 until 100) {
      val userid = i
      val username = "user" + i
      val name = "name" + i
      val age = random.nextInt(60)
      val professional = "professional" + random.nextInt(100)
      val city = "city" + random.nextInt(100)
      val sex = sexes(random.nextInt(2))
 
      rows += UserInfo(userid, username, name, age, professional, city, sex)
    }
 
    rows.toArray
  }
 
  /**
    * 模拟产品数据表
    *
    * @return
    */
  private def mockProductInfo(): Array[ProductInfo] = {
    val rows = ArrayBuffer[ProductInfo]()
    val random = new Random()
 
    val productStatus = Array(0, 1)
 
    // 随机产生 100 个产品信息
    for (i <- 0 until 100) {
      val productId = i
      val productName = "product" + i
      val extendInfo = "{\"product_status\": " + productStatus(random.nextInt(2)) + "}" // 注意这里是 json 串
 
      rows += ProductInfo(productId, productName, extendInfo)
    }
 
    rows.toArray
  }
 
  /**
    * 将 DataFrame 插入到 Hive 表中
    *
    * @param spark     SparkSQL 客户端
    * @param tableName 表名
    * @param dataDF    DataFrame
    */
  private def insertHive(spark: SparkSession, tableName: String, dataDF: DataFrame): Unit = {
    spark.sql("DROP TABLE IF EXISTS " + tableName)
    dataDF.write.saveAsTable(tableName)
  }
 
  val USER_VISIT_ACTION_TABLE = "user_visit_action"
  val USER_INFO_TABLE = "user_info"
  val PRODUCT_INFO_TABLE = "product_info"
 
  /**
    * 主入口方法
    *
    * @param args 启动参数
    */
  def main(args: Array[String]): Unit = {
 
    // 建立 Spark 配置
    val sparkConf = new SparkConf().setAppName("MockData").setMaster("local[*]")
 
    // 建立 Spark SQL 客户端
    val spark = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()
 
    // 模拟数据
    val userVisitActionData = this.mockUserVisitActionData()
    val userInfoData = this.mockUserInfo()
    val productInfoData = this.mockProductInfo()
 
    // 将模拟数据转换为 RDD
    val userVisitActionRdd = spark.sparkContext.makeRDD(userVisitActionData)
    val userInfoRdd = spark.sparkContext.makeRDD(userInfoData)
    val productInfoRdd = spark.sparkContext.makeRDD(productInfoData)
 
    // 加载 SparkSQL 的隐式转换支持
    import spark.implicits._
 
    // 将用户访问数据转换为 DF 保存到 Hive 表中
    val userVisitActionDF = userVisitActionRdd.toDF()
    insertHive(spark, USER_VISIT_ACTION_TABLE, userVisitActionDF)
 
    // 将用户信息数据转换为 DF 保存到 Hive 表中
    val userInfoDF = userInfoRdd.toDF()
    insertHive(spark, USER_INFO_TABLE, userInfoDF)
 
    // 将产品信息数据转换为 DF 保存到 Hive 表中
    val productInfoDF = productInfoRdd.toDF()
    insertHive(spark, PRODUCT_INFO_TABLE, productInfoDF)
 
    spark.close
  }
}

MockRealTimeData.scala

import java.util.Properties
 
import commons.conf.ConfigurationManager
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
 
import scala.collection.mutable.ArrayBuffer
import scala.util.Random
 
object MockRealTimeData {
 
  def main(args: Array[String]): Unit = {
    // 获取配置文件 commerce.properties 中的 Kafka 配置参数
    val broker = ConfigurationManager.config.getString("kafka.broker.list")
    val topic = ConfigurationManager.config.getString("kafka.topics")
 
    // 建立 Kafka 生产者
    val kafkaProducer = createKafkaProducer(broker)
 
    while (true) {
      // 随机产生实时数据并经过 Kafka 生产者发送到 Kafka 集群中
      for (item <- generateMockData()) {
        kafkaProducer.send(new ProducerRecord[String, String](topic, item))
      }
      Thread.sleep(5000)
    }
  }
 
  /**
    * 实时模拟数据的生成
    *
    * 时间点: 当前时间毫秒
    * userId: 0 - 99
    * 省份、城市 ID 相同: 1 - 9
    * adid: 0 - 19
    * ((0L,"北京","北京"),(1L,"上海","上海"),(2L,"南京","江苏省"),(3L,"广州","广东省"),(4L,"三亚","海南省"),(5L,"武汉","湖北省"),(6L,"长沙","湖南省"),(7L,"西安","陕西省"),(8L,"成都","四川省"),(9L,"哈尔滨","东北省"))
    *
    * 格式 ：timestamp province city userid adid
    *       某个时间点 某个省份 某个城市 某个用户 某个广告
    */
  def generateMockData(): Array[String] = {
    val array = ArrayBuffer[String]()
    val random = new Random()
 
    // 模拟实时数据：timestamp province city userid adid
    for (i <- 0 until 50) {
      val timestamp = System.currentTimeMillis()
      val province = random.nextInt(10)
      val city = province
      val adid = random.nextInt(20)
      val userid = random.nextInt(100)
 
      // 拼接实时数据
      array += timestamp + " " + province + " " + city + " " + userid + " " + adid
    }
 
    array.toArray
  }
 
  def createKafkaProducer(broker: String): KafkaProducer[String, String] = {
    // 建立配置对象
    val prop = new Properties()
    // 添加配置
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker)
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
 
    // 根据配置建立 Kafka 生产者
    new KafkaProducer[String, String](prop)
  }
}

4.2 commons 模块(公共模块)

新建一个模块 maven 工程 commons 做为子 maven 工程，引入依赖 pom.xml
pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>commerce</artifactId>
        <groupId>com.atguigu</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
 
    <artifactId>commons</artifactId>
 
    <dependencies>
        <dependency>
            <groupId>net.sf.json-lib</groupId>
            <artifactId>json-lib</artifactId>
            <version>2.4</version>
            <classifier>jdk15</classifier>
        </dependency>
 
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
            <version>2.4.2</version>
        </dependency>
 
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-configuration2</artifactId>
            <version>2.2</version>
        </dependency>
        <dependency>
            <groupId>commons-beanutils</groupId>
            <artifactId>commons-beanutils</artifactId>
            <version>1.9.3</version>
        </dependency>
 
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
        </dependency>
 
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.6</version>
        </dependency>
 
        <dependency>
            <groupId>joda-time</groupId>
            <artifactId>joda-time</artifactId>
            <version>2.9.9</version>
        </dependency>
 
    </dependencies>
 
    <build>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

类(包)名称

工具类名称

类(包)结构图

ConfigurationManager.scala

package commons.conf
 
import org.apache.commons.configuration2.{FileBasedConfiguration, PropertiesConfiguration}
import org.apache.commons.configuration2.builder.FileBasedConfigurationBuilder
import org.apache.commons.configuration2.builder.fluent.Parameters
 
/**
  * 配置工具类：新的读取配置文件信息的方式
  */
object ConfigurationManager {
  // 建立用于初始化配置生成器实例的参数对象
  private val params = new Parameters()
 
  // FileBasedConfigurationBuilder : 产生一个传入的类的实例对象
  // FileBasedConfiguration : 融合 FileBased 与 Configuration 的接口
  // PropertiesConfiguration : 从一个或者多个文件读取配置的标准配置加载器
  // configure() : 经过 params 实例初始化配置生成器
  // 向 FileBasedConfigurationBuilder() 中传入一个标准配置加载器类，生成一个加载器类的实例对象，而后经过 params 参数对其初始化
  private val builder = new FileBasedConfigurationBuilder[FileBasedConfiguration](classOf[PropertiesConfiguration])
    .configure(params.properties().setFileName("commerce.properties"))
 
  // 经过 getConfiguration 获取配置对象
  val config = builder.getConfiguration()
}

Constants.scala

package commons.constant
 
/**
  * 常量接口
  */
object Constants {
  /**
    * 项目配置相关的常量
    */
  val JDBC_DATASOURCE_SIZE = "jdbc.datasource.size"
  val JDBC_URL = "jdbc.url"
  val JDBC_USER = "jdbc.user"
  val JDBC_PASSWORD = "jdbc.password"
 
  val KAFKA_BROKERS = "kafka.broker.list"
  val KAFKA_TOPICS = "kafka.topics"
 
  /**
    * Spark 做业相关的常量
    */
  val SPARK_APP_NAME_SESSION = "UserVisitSessionAnalyzeSpark"
  val SPARK_APP_NAME_PAGE = "PageOneStepConvertRateSpark"
 
  /**
    * user_visit_action、user_info、product_info 表中字段对应的字段名常量
    */
  val FIELD_SESSION_ID = "sessionid"
  val FIELD_SEARCH_KEYWORDS = "searchKeywords"
  val FIELD_CLICK_CATEGORY_IDS = "clickCategoryIds"
  val FIELD_AGE = "age"
  val FIELD_PROFESSIONAL = "professional"
  val FIELD_CITY = "city"
  val FIELD_SEX = "sex"
  val FIELD_VISIT_LENGTH = "visitLength"
  val FIELD_STEP_LENGTH = "stepLength"
  val FIELD_START_TIME = "startTime"
  val FIELD_CLICK_COUNT = "clickCount"
  val FIELD_ORDER_COUNT = "orderCount"
  val FIELD_PAY_COUNT = "payCount"
  val FIELD_CATEGORY_ID = "categoryId"
 
  /**
    * Spark 累加器 Key 名称常量
    */
  val SESSION_COUNT = "session_count"
 
  val TIME_PERIOD_1s_3s = "1s_3s"
  val TIME_PERIOD_4s_6s = "4s_6s"
  val TIME_PERIOD_7s_9s = "7s_9s"
  val TIME_PERIOD_10s_30s = "10s_30s"
  val TIME_PERIOD_30s_60s = "30s_60s"
  val TIME_PERIOD_1m_3m = "1m_3m"
  val TIME_PERIOD_3m_10m = "3m_10m"
  val TIME_PERIOD_10m_30m = "10m_30m"
  val TIME_PERIOD_30m = "30m"
 
  val STEP_PERIOD_1_3 = "1_3"
  val STEP_PERIOD_4_6 = "4_6"
  val STEP_PERIOD_7_9 = "7_9"
  val STEP_PERIOD_10_30 = "10_30"
  val STEP_PERIOD_30_60 = "30_60"
  val STEP_PERIOD_60 = "60"
 
  /**
    * task.params.json 中限制条件对应的常量字段
    */
  val TASK_PARAMS = "task.params.json"
 
  val PARAM_START_DATE = "startDate"
  val PARAM_END_DATE = "endDate"
  val PARAM_START_AGE = "startAge"
  val PARAM_END_AGE = "endAge"
  val PARAM_PROFESSIONALS = "professionals"
  val PARAM_CITIES = "cities"
  val PARAM_SEX = "sex"
  val PARAM_KEYWORDS = "keywords"
  val PARAM_CATEGORY_IDS = "categoryIds"
  val PARAM_TARGET_PAGE_FLOW = "targetPageFlow"
}

DataModel.scala

package commons.model
 
//***************** 输入表 *********************
 
/**
  * 用户访问动做表
  *
  * @param date               用户点击行为的日期
  * @param user_id            用户的 ID
  * @param session_id         Session 的 ID
  * @param page_id            某个页面的 ID
  * @param action_time        点击行为的时间点
  * @param search_keyword     用户搜索的关键词
  * @param click_category_id  某一个商品品类的 ID
  * @param click_product_id   某一个商品的 ID
  * @param order_category_ids 一次订单中全部品类的 ID 集合
  * @param order_product_ids  一次订单中全部商品的 ID 集合
  * @param pay_category_ids   一次支付中全部品类的 ID 集合
  * @param pay_product_ids    一次支付中全部商品的 ID 集合
  * @param city_id            城市 ID
  */
case class UserVisitAction(date: String,
                           user_id: Long,
                           session_id: String,
                           page_id: Long,
                           action_time: String,
                           search_keyword: String,
                           click_category_id: Long,
                           click_product_id: Long,
                           order_category_ids: String,
                           order_product_ids: String,
                           pay_category_ids: String,
                           pay_product_ids: String,
                           city_id: Long)
 
/**
  * 用户信息表
  *
  * @param user_id      用户的 ID
  * @param username     用户的名称
  * @param name         用户的名字
  * @param age          用户的年龄
  * @param professional 用户的职业
  * @param city         用户所在的城市
  * @param sex          用户的性别
  */
case class UserInfo(user_id: Long,
                    username: String,
                    name: String,
                    age: Int,
                    professional: String,
                    city: String,
                    sex: String)
 
/**
  * 产品表
  *
  * @param product_id   商品的 ID
  * @param product_name 商品的名称
  * @param extend_info  商品额外的信息
  */
case class ProductInfo(product_id: Long,
                       product_name: String,
                       extend_info: String)

PooledMySqlClientFactory.scala

package commons.pool
 
import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
 
import commons.conf.ConfigurationManager
import commons.constant.Constants
import org.apache.commons.pool2.impl.{DefaultPooledObject, GenericObjectPool, GenericObjectPoolConfig}
import org.apache.commons.pool2.{BasePooledObjectFactory, PooledObject}
 
// 建立用于处理 MySQL 查询结果的类的抽象接口
trait QueryCallback {
  def process(rs: ResultSet)
}
 
/**
  * MySQL 客户端代理对象
  *
  * @param jdbcUrl      MySQL URL
  * @param jdbcUser     MySQL 用户
  * @param jdbcPassword MySQL 密码
  * @param client       默认客户端实现
  */
case class MySqlProxy(jdbcUrl: String, jdbcUser: String, jdbcPassword: String, client: Option[Connection] = None) {
  // 获取客户端链接对象
  private val mysqlClient = client getOrElse {
    DriverManager.getConnection(jdbcUrl, jdbcUser, jdbcPassword)
  }
 
  /**
    * 执行增删改 SQL 语句
    *
    * @param sql
    * @param params
    * @return 影响的行数
    */
  def executeUpdate(sql: String, params: Array[Any]): Int = {
    var rtn = 0
    var pstmt: PreparedStatement = null
 
    try {
      // 第一步：关闭自动提交
      mysqlClient.setAutoCommit(false)
 
      // 第二步：根据传入的 sql 语句建立 prepareStatement
      pstmt = mysqlClient.prepareStatement(sql)
 
      // 第三步：为 prepareStatement 中的每一个参数填写数值
      if (params != null && params.length > 0) {
        for (i <- 0 until params.length) {
          pstmt.setObject(i + 1, params(i))
        }
      }
 
      // 第四步：执行增删改操做
      rtn = pstmt.executeUpdate()
 
      // 第五步：手动提交
      mysqlClient.commit()
    } catch {
      case e: Exception => e.printStackTrace
    }
 
    rtn
  }
 
  /**
    * 执行查询 SQL 语句
    *
    * @param sql
    * @param params
    */
  def executeQuery(sql: String, params: Array[Any], queryCallback: QueryCallback) {
    var pstmt: PreparedStatement = null
    var rs: ResultSet = null
 
    try {
      // 第一步：根据传入的 sql 语句建立 prepareStatement
      pstmt = mysqlClient.prepareStatement(sql)
 
      // 第二步：为 prepareStatement 中的每一个参数填写数值
      if (params != null && params.length > 0) {
        for (i <- 0 until params.length) {
          pstmt.setObject(i + 1, params(i))
        }
      }
 
      // 第三步：执行查询操做
      rs = pstmt.executeQuery()
 
      // 第四步：处理查询后的结果
      queryCallback.process(rs)
    } catch {
      case e: Exception => e.printStackTrace
    }
  }
 
  /**
    * 批量执行 SQL 语句
    *
    * @param sql
    * @param paramsList
    * @return 每条SQL语句影响的行数
    */
  def executeBatch(sql: String, paramsList: Array[Array[Any]]): Array[Int] = {
    var rtn: Array[Int] = null
    var pstmt: PreparedStatement = null
    try {
      // 第一步：关闭自动提交
      mysqlClient.setAutoCommit(false)
 
      pstmt = mysqlClient.prepareStatement(sql)
 
      // 第二步：为 prepareStatement 中的每一个参数填写数值
      if (paramsList != null && paramsList.length > 0) {
        for (params <- paramsList) {
          for (i <- 0 until params.length) {
            pstmt.setObject(i + 1, params(i))
          }
          pstmt.addBatch()
        }
      }
 
      // 第三步：执行批量的 SQL 语句
      rtn = pstmt.executeBatch()
 
      // 第四步：手动提交
      mysqlClient.commit()
    } catch {
      case e: Exception => e.printStackTrace
    }
 
    rtn
  }
 
  // 关闭 MySQL 客户端
  def shutdown(): Unit = mysqlClient.close()
}
 
/**
  * 扩展知识：将 MySqlProxy 实例视为对象，MySqlProxy 实例的建立使用对象池进行维护
  *
  * 建立自定义工厂类，继承 BasePooledObjectFactory 工厂类，负责对象的建立、包装和销毁
  *
  * @param jdbcUrl
  * @param jdbcUser
  * @param jdbcPassword
  * @param client
  */
class PooledMySqlClientFactory(jdbcUrl: String, jdbcUser: String, jdbcPassword: String, client: Option[Connection] = None)
  extends BasePooledObjectFactory[MySqlProxy] with Serializable {
  // 用于池来建立对象
  override def create(): MySqlProxy = MySqlProxy(jdbcUrl, jdbcUser, jdbcPassword, client)
 
  // 用于池来包装对象
  override def wrap(obj: MySqlProxy): PooledObject[MySqlProxy] = new DefaultPooledObject(obj)
 
  // 用于池来销毁对象
  override def destroyObject(p: PooledObject[MySqlProxy]): Unit = {
    p.getObject.shutdown()
    super.destroyObject(p)
  }
}
 
/**
  * 建立 MySQL 池工具类
  */
object CreateMySqlPool {
  // 加载 JDBC 驱动，只须要一次
  Class.forName("com.mysql.jdbc.Driver")
 
  // 在 org.apache.commons.pool2.impl 中预设了三个能够直接使用的对象池：GenericObjectPool、GenericKeyedObjectPool 和 SoftReferenceObjectPool
  // 建立 genericObjectPool 为 GenericObjectPool
  // GenericObjectPool 的特色是能够设置对象池中的对象特征，包括 LIFO 方式、最大空闲数、最小空闲数、是否有效性检查等等
  private var genericObjectPool: GenericObjectPool[MySqlProxy] = null
 
  // 伴生对象经过 apply 完成对象的建立
  def apply(): GenericObjectPool[MySqlProxy] = {
    // 单例模式
    if (this.genericObjectPool == null) {
      this.synchronized {
        // 获取 MySQL 配置参数
        val jdbcUrl = ConfigurationManager.config.getString(Constants.JDBC_URL)
        val jdbcUser = ConfigurationManager.config.getString(Constants.JDBC_USER)
        val jdbcPassword = ConfigurationManager.config.getString(Constants.JDBC_PASSWORD)
        val size = ConfigurationManager.config.getInt(Constants.JDBC_DATASOURCE_SIZE)
 
        val pooledFactory = new PooledMySqlClientFactory(jdbcUrl, jdbcUser, jdbcPassword)
        val poolConfig = {
          // 建立标准对象池配置类的实例
          val c = new GenericObjectPoolConfig
          // 设置配置对象参数
          // 设置最大对象数
          c.setMaxTotal(size)
          // 设置最大空闲对象数
          c.setMaxIdle(size)
          c
        }
 
        // 对象池的建立须要工厂类和配置类
        // 返回一个 GenericObjectPool 对象池
        this.genericObjectPool = new GenericObjectPool[MySqlProxy](pooledFactory, poolConfig)
      }
    }
 
    genericObjectPool
  }
}

Utils.scala

package commons.utils
 
import java.util.Date
 
import net.sf.json.JSONObject
import org.joda.time.DateTime
import org.joda.time.format.DateTimeFormat
 
import scala.collection.mutable
 
/**
  * 日期时间工具类
  * 使用 joda 实现，若是使用 Java 提供的 Date 会存在线程安全问题
  */
object DateUtils {
  val DATE_FORMAT = DateTimeFormat.forPattern("yyyy-MM-dd")
  val TIME_FORMAT = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")
 
  val DATE_KEY_FORMAT = DateTimeFormat.forPattern("yyyyMMdd")
  val DATE_TIME_FORMAT = DateTimeFormat.forPattern("yyyyMMddHHmm")
 
  /**
    * 判断一个时间是否在另外一个时间以前
    *
    * @param time1 第一个时间
    * @param time2 第二个时间
    * @return 判断结果
    */
  def before(time1: String, time2: String): Boolean = {
    if (TIME_FORMAT.parseDateTime(time1).isBefore(TIME_FORMAT.parseDateTime(time2))) {
      return true
    }
    false
  }
 
  /**
    * 判断一个时间是否在另外一个时间以后
    *
    * @param time1 第一个时间
    * @param time2 第二个时间
    * @return 判断结果
    */
  def after(time1: String, time2: String): Boolean = {
    if (TIME_FORMAT.parseDateTime(time1).isAfter(TIME_FORMAT.parseDateTime(time2))) {
      return true
    }
    false
  }
 
  /**
    * 计算时间差值（单位为秒）
    *
    * @param time1 时间1
    * @param time2 时间2
    * @return 差值
    */
  def minus(time1: String, time2: String): Int = {
    return (TIME_FORMAT.parseDateTime(time1).getMillis - TIME_FORMAT.parseDateTime(time2).getMillis) / 1000 toInt
  }
 
  /**
    * 获取年月日和小时
    *
    * @param datetime 时间（yyyy-MM-dd HH:mm:ss）
    * @return 结果（yyyy-MM-dd_HH）
    */
  def getDateHour(datetime: String): String = {
    val date = datetime.split(" ")(0)
    val hourMinuteSecond = datetime.split(" ")(1)
    val hour = hourMinuteSecond.split(":")(0)
    date + "_" + hour
  }
 
  /**
    * 获取当天日期（yyyy-MM-dd）
    *
    * @return 当天日期
    */
  def getTodayDate(): String = {
    DateTime.now().toString(DATE_FORMAT)
  }
 
  /**
    * 获取昨天的日期（yyyy-MM-dd）
    *
    * @return 昨天的日期
    */
  def getYesterdayDate(): String = {
    DateTime.now().minusDays(1).toString(DATE_FORMAT)
  }
 
  /**
    * 格式化日期（yyyy-MM-dd）
    *
    * @param date Date对象
    * @return 格式化后的日期
    */
  def formatDate(date: Date): String = {
    new DateTime(date).toString(DATE_FORMAT)
  }
 
  /**
    * 格式化时间（yyyy-MM-dd HH:mm:ss）
    *
    * @param date Date对象
    * @return 格式化后的时间
    */
  def formatTime(date: Date): String = {
    new DateTime(date).toString(TIME_FORMAT)
  }
 
  /**
    * 解析时间字符串
    *
    * @param time 时间字符串
    * @return Date
    */
  def parseTime(time: String): Date = {
    TIME_FORMAT.parseDateTime(time).toDate
  }
 
  def main(args: Array[String]): Unit = {
    print(DateUtils.parseTime("2017-10-31 20:27:53")) // Tue Oct 31 20:27:53 CST 2017
  }
 
  /**
    * 格式化日期 key
    * yyyyMMdd
    *
    * @param date
    * @return
    */
  def formatDateKey(date: Date): String = {
    new DateTime(date).toString(DATE_KEY_FORMAT)
  }
 
  /**
    * 解析日期 key
    *
    * @return
    */
  def parseDateKey(datekey: String): Date = {
    DATE_KEY_FORMAT.parseDateTime(datekey).toDate
  }
 
  /**
    * 格式化时间，保留到分钟级别
    * yyyyMMddHHmm
    *
    * @param date
    * @return
    */
  def formatTimeMinute(date: Date): String = {
    new DateTime(date).toString(DATE_TIME_FORMAT)
  }
}
 
/**
  * 数字格式化工具类
  */
object NumberUtils {
  /**
    * 格式化小数
    *
    * @param scale 四舍五入的位数
    * @return 格式化小数
    */
  def formatDouble(num: Double, scale: Int): Double = {
    val bd = BigDecimal(num)
    bd.setScale(scale, BigDecimal.RoundingMode.HALF_UP).doubleValue()
  }
}
 
/**
  * 参数工具类
  */
object ParamUtils {
  /**
    * 从 JSON 对象中提取参数
    *
    * @param jsonObject JSON对象
    * @return 参数
    */
  def getParam(jsonObject: JSONObject, field: String): String = {
    jsonObject.getString(field)
  }
}
 
/**
  * 字符串工具类
  *
  */
object StringUtils {
  /**
    * 判断字符串是否为空
    *
    * @param str 字符串
    * @return 是否为空
    */
  def isEmpty(str: String): Boolean = {
    str == null || "".equals(str)
  }
 
  /**
    * 判断字符串是否不为空
    *
    * @param str 字符串
    * @return 是否不为空
    */
  def isNotEmpty(str: String): Boolean = {
    str != null && !"".equals(str)
  }
 
  /**
    * 截断字符串两侧的逗号
    *
    * @param str 字符串
    * @return 字符串
    */
  def trimComma(str: String): String = {
    var result = ""
    if (str.startsWith(",")) {
      result = str.substring(1)
    }
    if (str.endsWith(",")) {
      result = str.substring(0, str.length() - 1)
    }
    result
  }
 
  /**
    * 补全两位数字
    *
    * @param str
    * @return
    */
  def fulfuill(str: String): String = {
    if (str.length() == 2) {
      str
    } else {
      "0" + str
    }
  }
 
  /**
    * 从拼接的字符串中提取字段
    *
    * @param str       字符串
    * @param delimiter 分隔符
    * @param field     字段
    * @return 字段值
    */
  def getFieldFromConcatString(str: String, delimiter: String, field: String): String = {
    try {
      val fields = str.split(delimiter);
      for (concatField <- fields) {
        if (concatField.split("=").length == 2) {
          val fieldName = concatField.split("=")(0)
          val fieldValue = concatField.split("=")(1)
          if (fieldName.equals(field)) {
            return fieldValue
          }
        }
      }
    } catch {
      case e: Exception => e.printStackTrace()
    }
 
    null
  }
 
  /**
    * 从拼接的字符串中给字段设置值
    *
    * @param str           字符串
    * @param delimiter     分隔符
    * @param field         字段名
    * @param newFieldValue 新的field值
    * @return 字段值
    */
  def setFieldInConcatString(str: String, delimiter: String, field: String, newFieldValue: String): String = {
    val fieldsMap = new mutable.HashMap[String, String]()
 
    for (fileds <- str.split(delimiter)) {
      val arra = fileds.split("=")
      if (arra(0).compareTo(field) == 0)
        fieldsMap += (field -> newFieldValue)
      else
        fieldsMap += (arra(0) -> arra(1))
    }
 
    fieldsMap.map(item => item._1 + "=" + item._2).mkString(delimiter)
  }
}
 
/**
  * 校验工具类
  */
object ValidUtils {
  /**
    * 校验数据中的指定字段，是否在指定范围内（范围区间）
    *
    * @param data            数据
    * @param dataField       数据字段
    * @param parameter       参数
    * @param startParamField 起始参数字段
    * @param endParamField   结束参数字段
    * @return 校验结果
    */
  def between(data: String, dataField: String, parameter: String, startParamField: String, endParamField: String): Boolean = {
    val startParamFieldStr = StringUtils.getFieldFromConcatString(parameter, "\\|", startParamField)
    val endParamFieldStr = StringUtils.getFieldFromConcatString(parameter, "\\|", endParamField)
    if (startParamFieldStr == null || endParamFieldStr == null) {
      return true
    }
 
    val startParamFieldValue = startParamFieldStr.toInt
    val endParamFieldValue = endParamFieldStr.toInt
 
    val dataFieldStr = StringUtils.getFieldFromConcatString(data, "\\|", dataField)
    if (dataFieldStr != null) {
      val dataFieldValue = dataFieldStr.toInt
      if (dataFieldValue >= startParamFieldValue && dataFieldValue <= endParamFieldValue) {
        return true
      } else {
        return false
      }
    }
 
    false
  }
 
  /**
    * 校验数据中的指定字段，是否有值与参数字段的值相同（多选一）
    *
    * @param data       数据
    * @param dataField  数据字段
    * @param parameter  参数
    * @param paramField 参数字段
    * @return 校验结果
    */
  def in(data: String, dataField: String, parameter: String, paramField: String): Boolean = {
    val paramFieldValue = StringUtils.getFieldFromConcatString(parameter, "\\|", paramField)
    if (paramFieldValue == null) {
      return true
    }
    val paramFieldValueSplited = paramFieldValue.split(",")
 
    val dataFieldValue = StringUtils.getFieldFromConcatString(data, "\\|", dataField)
    if (dataFieldValue != null && dataFieldValue != "-1") {
      val dataFieldValueSplited = dataFieldValue.split(",")
 
      for (singleDataFieldValue <- dataFieldValueSplited) {
        for (singleParamFieldValue <- paramFieldValueSplited) {
          if (singleDataFieldValue.compareTo(singleParamFieldValue) == 0) {
            return true
          }
        }
      }
    }
 
    false
  }
 
  /**
    * 校验数据中的指定字段，是否在指定范围内（二选一）
    *
    * @param data       数据
    * @param dataField  数据字段
    * @param parameter  参数
    * @param paramField 参数字段
    * @return 校验结果
    */
  def equal(data: String, dataField: String, parameter: String, paramField: String): Boolean = {
    val paramFieldValue = StringUtils.getFieldFromConcatString(parameter, "\\|", paramField)
    if (paramFieldValue == null) {
      return true
    }
 
    val dataFieldValue = StringUtils.getFieldFromConcatString(data, "\\|", dataField)
    if (dataFieldValue != null) {
      if (dataFieldValue.compareTo(paramFieldValue) == 0) {
        return true
      }
    }
 
    false
  }
}

---

commerce.properties

# jbdc 配置
jdbc.datasource.size=10
jdbc.url=jdbc:mysql://localhost:3306/commerce?useUnicode=true&characterEncoding=utf8
jdbc.user=root
jdbc.password=root
 
# 筛选条件的配置
# 可使用的属性以下：
#      startDate:       格式： yyyy-MM-DD   [必选]
#      endDate:         格式： yyyy-MM-DD   [必选]
#      startAge:        范围： 0 - 59
#      endAge:          范围： 0 - 59
#      professionals:   范围：professionals[0 - 99]
#      cities:          0 - 99  ((0,"北京","华北"),(1,"上海","华东"),(2,"南京","华东"),(3,"广州","华南"),(4,"三亚","华南"),(5,"武汉","华中"),(6,"长沙","华中"),(7,"西安","西北"),(8,"成都","西南"),(9,"哈尔滨","东北"),...)
#      sex:             范围： 0 - 1
#      keywords:        范围： ("火锅", "蛋糕", "重庆辣子鸡", "重庆小面", "呷哺呷哺", "新辣道鱼火锅", "国贸大厦", "太古商场", "日本料理", "温泉")
#      categoryIds:     0 - 99，以逗号分隔
#      targetPageFlow:  0 - 99， 以逗号分隔
task.params.json={\
  startDate:"2019-06-01", \
  endDate:"2019-06-30", \
  startAge: 20, \
  endAge: 50, \
  professionals: "",  \
  cities: "", \
  sex:"", \
  keywords:"", \
  categoryIds:"", \
  targetPageFlow:"1,2,3,4,5,6,7"}
 
# Kafka 配置
kafka.broker.list=hadoop102:9092,hadoop103:9092,hadoop104:9092
kafka.topics=AdRealTimeLog1

log4j.properties

log4j.rootLogger=info, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS}  %5p --- [%50t]  %-80c(line:%5L)  :  %m%n
 
log4j.appender.R=org.apache.log4j.RollingFileAppender
log4j.appender.R.File=../log/agent.log
log4j.appender.R.MaxFileSize=1024KB
log4j.appender.R.MaxBackupIndex=1
 
log4j.appender.R.layout=org.apache.log4j.PatternLayout
log4j.appender.R.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS}  %5p --- [%50t]  %-80c(line:%6L)  :  %m%n

4.3 analyse 模块(数据分析模块)

新建一个模块 maven 工程 analyse 做为子 maven 工程，删除掉 src 目录，引入依赖 pom.xml，添加对 scala 框架的支持。
注意：在该子模块中有不少子模块。即具体需求实现的模块。
pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>commerce</artifactId>
        <groupId>com.atguigu</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
 
    <artifactId>analyse</artifactId>
 
 
</project>

analyse 模块是需求的具体实现模块， 咱们将会在第 5 章中进行详细解析。

第5章 需求解析

5.1 需求一：Session 各范围访问步长、访问时长占比统计

5.1.1 需求解析

  需求一：要统计出符合筛选条件的 session 中，访问时长在 1s~3s、4s~6s、7s~9s、10s~30s、30s~60s、1m~3m、3m~10m、10m~30m、30m 以上各个范围内的 session 占比；访问步长在 1~三、4~六、7~九、10~30、30~60、60 以上各个范围内的 session 占比，并将结果保存到 MySQL 数据库中。
  在计算以前须要根据查询条件筛选 session，查询条件好比搜索过某些关键词的用户、访问时间在某个时间段内的用户、年龄在某个范围内的用户、职业在某个范围内的用户、所在某个城市的用户，发起的 session。找到对应的这些用户的 session，并进行统计，之因此须要有筛选主要是可让使用者，对感兴趣的和关系的用户群体，进行后续各类复杂业务逻辑的统计和分析，那么拿到的结果数据，就是只是针对特殊用户群体的分析结果；而不是对全部用户进行分析的泛泛的分析结果。好比说，如今某个企业高层，就是想看到用户群体中，28~35 岁的老师职业的群体，对应的一些统计和分析的结果数据，从而辅助高管进行公司战略上的决策制定。
  session 访问时长，也就是说一个 session 对应的开始的 action 到结束的 action 之间的时间范围；还有，就是访问步长，指的是，一个 session 执行期间内，依次点击过多少个页面，好比说，一次 session 维持了 1 分钟，那么访问时长就是 1m，而后在这 1 分钟内，点击了 10 个页面，那么 session 的访问步长，就是 10。
  好比说，符合第一步筛选出来的 session 的数量大概是有 1000 万个。那么里面，咱们要计算出，访问时长在 1s~3s 内的 session 的数量，并除以符合条件的总 session 数量（好比 1000 万），好比是 100 万/1000 万，那么 1s~3s 内的 session 占比就是 10%。依次类推，这里说的统计，就是这个意思。
  这个功能可让人从全局的角度看到，符合某些条件的用户群体使用咱们的产品的一些习惯。好比大多数人，究竟是会在产品中停留多长时间，大多数人，会在一次使用产品的过程当中，访问多少个页面。那么对于使用者来讲， 有一个全局和清晰的认识。

5.1.2 数据源解析

5.1.3 数据结构解析

一、UserVisitAction 样例类

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

二、UserInfo 样例类

/**
  * 用户信息表
  *
  * @param user_id      用户的 ID
  * @param username     用户的名称
  * @param name         用户的名字
  * @param age          用户的年龄
  * @param professional 用户的职业
  * @param city         用户所在的城市
  * @param sex          用户的性别
  */
case class UserInfo(user_id: Long,
                    username: String,
                    name: String,
                    age: Int,
                    professional: String,
                    city: String,
                    sex: String
                   )

为何联立用户表？
  用户表中记录了用户详细的我的信息，包括年龄、职业、城市、性别等，在实际的业务场景中，咱们可能会在一段时间关注某一个群体的用户的行为，好比在某一段时间关注北京的白领们的购物行为，那么咱们就能够经过联立用户表，让咱们的统计数据中具备用户属性，而后根据用户属性对统计信息进行过滤，将不属于咱们所关注的用户群体的用户所产生的行为数据过滤掉，这样就能够实现对指定人群的精准分析。

5.1.4 需求实现简要流程

5.1.5 需求实现详细流程

5.1.6 MySQL 存储结构解析

MySQL 写入数据格式
session_aggr_stat

-- ----------------------------
--  Table structure for `session_aggr_stat`
-- ----------------------------
DROP TABLE IF EXISTS `session_aggr_stat`;
CREATE TABLE `session_aggr_stat` (
  `taskid` varchar(255) DEFAULT NULL,
  `session_count` int(11) DEFAULT NULL,
  `visit_length_1s_3s_ratio` double DEFAULT NULL,
  `visit_length_4s_6s_ratio` double DEFAULT NULL,
  `visit_length_7s_9s_ratio` double DEFAULT NULL,
  `visit_length_10s_30s_ratio` double DEFAULT NULL,
  `visit_length_30s_60s_ratio` double DEFAULT NULL,
  `visit_length_1m_3m_ratio` double DEFAULT NULL,
  `visit_length_3m_10m_ratio` double DEFAULT NULL,
  `visit_length_10m_30m_ratio` double DEFAULT NULL,
  `visit_length_30m_ratio` double DEFAULT NULL,
  `step_length_1_3_ratio` double DEFAULT NULL,
  `step_length_4_6_ratio` double DEFAULT NULL,
  `step_length_7_9_ratio` double DEFAULT NULL,
  `step_length_10_30_ratio` double DEFAULT NULL,
  `step_length_30_60_ratio` double DEFAULT NULL,
  `step_length_60_ratio` double DEFAULT NULL,
  KEY `idx_task_id` (`taskid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

5.1.7 代码解析

在模块 analyse 新建一个模块 session，引入 pom 文件，修改 src 目录名称为 scala，同时添加 scala 框架的支持。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>analyse</artifactId>
        <groupId>com.atguigu</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
 
    <artifactId>session</artifactId>
 
    <dependencies>
        <dependency>
            <groupId>com.atguigu</groupId>
            <artifactId>commons</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <!-- Spark 的依赖引入 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
        </dependency>
        <!-- 引入 Scala -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
        </dependency>
    </dependencies>
 
    <build>
        <plugins>
            <plugin>
                <!-- scala-maven-plugin 插件用于在任意的 maven 项目中对 scala 代码进行编译/测试/运行/文档化 -->
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.atguigu.session.UserVisitSessionAnalyze</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

代码实现示例以下：
SessionStat.scala

package com.atguigu.session
 
import java.util.{Date, UUID}
 
import commons.conf.ConfigurationManager
import commons.constant.Constants
import commons.model.{UserInfo, UserVisitAction}
import commons.utils._
import net.sf.json.JSONObject
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SaveMode, SparkSession}
 
import scala.collection.mutable
 
object SessionStat {
 
  def main(args: Array[String]): Unit = {
    // 获取过滤条件，【为了方便，直接从配置文件中获取，企业中会从一个调度平台获取】
    val jsonStr = ConfigurationManager.config.getString(Constants.TASK_PARAMS)
    // 获取过滤条件对应的 JsonObject 对象
    val taskParam = JSONObject.fromObject(jsonStr)
 
    // 建立全局惟一的主键，每次执行 main 函数都会生成一个独一无二的 taskUUID，来区分不一样任务，做为写入 MySQL 数据库中那张表的主键
    val taskUUID = UUID.randomUUID().toString
 
    // 建立 sparkConf
    val sparkConf = new SparkConf().setAppName("session").setMaster("local[*]")
 
    // 建立 sparkSession（包含 sparkContext）
    val sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()
 
    // 获取原始的动做表数据（带有过滤条件）
    // actionRDD: RDD[UserVisitAction]
    val actionRDD = getOriActionRDD(sparkSession, taskParam)
 
    // 将用户行为信息转换为 K-V 结构，sessionId2ActionRDD: RDD[(sessionId, UserVisitAction)]
    val sessionId2ActionRDD = actionRDD.map(item => (item.session_id, item))
 
    // session2GroupActionRDD: RDD[(sessionId, Iterable[UserVisitAction])]
    val session2GroupActionRDD = sessionId2ActionRDD.groupByKey() // 把同一个 sessionId 的数据聚合到一块儿，获得斧子形数据
 
    // 将数据进行内存缓存
    session2GroupActionRDD.cache()
 
    // sessionId2FullAggrInfoRDD: RDD[(sessionId, fullAggrInfo)]
    val sessionId2FullAggrInfoRDD = getSessionFullAggrInfo(sparkSession, session2GroupActionRDD)
 
    // 建立自定义累加器对象
    val sessionStatisticAccumulator = new SessionStatisticAccumulator
 
    // 在 sparkSession 中注册自定义累加器，这样后面就能够用了
    sparkSession.sparkContext.register(sessionStatisticAccumulator)
 
    // 根据过滤条件对 sessionId2FullAggrInfoRDD 进行过滤操做，即过滤掉不符合条件的数据，并根据自定义累加器 统计不一样范围的 访问时长 和 访问步长 的 session 个数 以及 总的 session 个数
    val seeionId2FilterRDD = getSessionFilterRDD(taskParam, sessionId2FullAggrInfoRDD, sessionStatisticAccumulator)
 
    // 必须引入任意一个 action 的算子，才能启动
    seeionId2FilterRDD.foreach(println(_))
 
    // 计算各个 session 的占比
    getSessionRatio(sparkSession,taskUUID, sessionStatisticAccumulator.value)
  }
 
  def getSessionRatio(sparkSession: SparkSession, taskUUID: String, value: mutable.HashMap[String, Int]): Unit = {
    val session_count = value.getOrElse(Constants.SESSION_COUNT, 1).toDouble
 
    // 先获取各个值
    val visit_length_1s_3s = value.getOrElse(Constants.TIME_PERIOD_1s_3s, 0)
    val visit_length_4s_6s = value.getOrElse(Constants.TIME_PERIOD_4s_6s, 0)
    val visit_length_7s_9s = value.getOrElse(Constants.TIME_PERIOD_7s_9s, 0)
    val visit_length_10s_30s = value.getOrElse(Constants.TIME_PERIOD_10s_30s, 0)
    val visit_length_30s_60s = value.getOrElse(Constants.TIME_PERIOD_30s_60s, 0)
    val visit_length_1m_3m = value.getOrElse(Constants.TIME_PERIOD_1m_3m, 0)
    val visit_length_3m_10m = value.getOrElse(Constants.TIME_PERIOD_3m_10m, 0)
    val visit_length_10m_30m = value.getOrElse(Constants.TIME_PERIOD_10m_30m, 0)
    val visit_length_30m = value.getOrElse(Constants.TIME_PERIOD_30m, 0)
 
    val step_length_1_3 = value.getOrElse(Constants.STEP_PERIOD_1_3, 0)
    val step_length_4_6 = value.getOrElse(Constants.STEP_PERIOD_4_6, 0)
    val step_length_7_9 = value.getOrElse(Constants.STEP_PERIOD_7_9, 0)
    val step_length_10_30 = value.getOrElse(Constants.STEP_PERIOD_10_30, 0)
    val step_length_30_60 = value.getOrElse(Constants.STEP_PERIOD_30_60, 0)
    val step_length_60 = value.getOrElse(Constants.STEP_PERIOD_60, 0)
 
    // 计算比例
    val visit_length_1s_3s_ratio = NumberUtils.formatDouble(visit_length_1s_3s / session_count, 2)
    val visit_length_4s_6s_ratio = NumberUtils.formatDouble(visit_length_4s_6s / session_count, 2)
    val visit_length_7s_9s_ratio = NumberUtils.formatDouble(visit_length_7s_9s / session_count, 2)
    val visit_length_10s_30s_ratio = NumberUtils.formatDouble(visit_length_10s_30s / session_count, 2)
    val visit_length_30s_60s_ratio = NumberUtils.formatDouble(visit_length_30s_60s / session_count, 2)
    val visit_length_1m_3m_ratio = NumberUtils.formatDouble(visit_length_1m_3m / session_count, 2)
    val visit_length_3m_10m_ratio = NumberUtils.formatDouble(visit_length_3m_10m / session_count, 2)
    val visit_length_10m_30m_ratio = NumberUtils.formatDouble(visit_length_10m_30m / session_count, 2)
    val visit_length_30m_ratio = NumberUtils.formatDouble(visit_length_30m / session_count, 2)
 
    val step_length_1_3_ratio = NumberUtils.formatDouble(step_length_1_3 / session_count, 2)
    val step_length_4_6_ratio = NumberUtils.formatDouble(step_length_4_6 / session_count, 2)
    val step_length_7_9_ratio = NumberUtils.formatDouble(step_length_7_9 / session_count, 2)
    val step_length_10_30_ratio = NumberUtils.formatDouble(step_length_10_30 / session_count, 2)
    val step_length_30_60_ratio = NumberUtils.formatDouble(step_length_30_60 / session_count, 2)
    val step_length_60_ratio = NumberUtils.formatDouble(step_length_60 / session_count, 2)
 
    // 封装数据
    val stat = SessionAggrStat(taskUUID, session_count.toInt,
      visit_length_1s_3s_ratio, visit_length_4s_6s_ratio, visit_length_7s_9s_ratio,
      visit_length_10s_30s_ratio, visit_length_30s_60s_ratio, visit_length_1m_3m_ratio,
      visit_length_3m_10m_ratio, visit_length_10m_30m_ratio, visit_length_30m_ratio,
      step_length_1_3_ratio, step_length_4_6_ratio, step_length_7_9_ratio,
      step_length_10_30_ratio, step_length_30_60_ratio, step_length_60_ratio)
 
    // 样例类实例 -> 数组 -> RDD
    val sessionRatioRDD = sparkSession.sparkContext.makeRDD(Array(stat))
 
    // 写入 MySQL 数据库中
    import sparkSession.implicits._
    sessionRatioRDD.toDF().write
      .format("jdbc")
      .option("url", ConfigurationManager.config.getString(Constants.JDBC_URL))
      .option("user", ConfigurationManager.config.getString(Constants.JDBC_USER))
      .option("password", ConfigurationManager.config.getString(Constants.JDBC_PASSWORD))
      .option("dbtable", "session_aggr_stat")
      .mode(SaveMode.Append) // 表存在就追加，表不存在就新建
      .save()
  }
 
  def calculateVisitLength(visitLength: Long, sessionStatisticAccumulator: SessionStatisticAccumulator) = {
    if (visitLength >= 1 && visitLength <= 3) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_1s_3s)
    } else if (visitLength >= 4 && visitLength <= 6) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_4s_6s)
    } else if (visitLength >= 7 && visitLength <= 9) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_7s_9s)
    } else if (visitLength >= 10 && visitLength <= 30) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_10s_30s)
    } else if (visitLength > 30 && visitLength <= 60) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_30s_60s)
    } else if (visitLength > 60 && visitLength <= 180) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_1m_3m)
    } else if (visitLength > 180 && visitLength <= 600) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_3m_10m)
    } else if (visitLength > 600 && visitLength <= 1800) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_10m_30m)
    } else if (visitLength > 1800) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_30m)
    }
  }
 
  def calculateStepLength(stepLength: Long, sessionStatisticAccumulator: SessionStatisticAccumulator) = {
    if (stepLength >= 1 && stepLength <= 3) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_1_3)
    } else if (stepLength >= 4 && stepLength <= 6) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_4_6)
    } else if (stepLength >= 7 && stepLength <= 9) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_7_9)
    } else if (stepLength >= 10 && stepLength <= 30) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_10_30)
    } else if (stepLength > 30 && stepLength <= 60) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_30_60)
    } else if (stepLength > 60) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_60)
    }
  }
 
  def getSessionFilterRDD(taskParam: JSONObject,
                          sessionId2FullAggrInfoRDD: RDD[(String, String)],
                          sessionStatisticAccumulator: SessionStatisticAccumulator) = {
    // 先获取所用到的过滤条件：
    val startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE)
    val endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE)
    val professionals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS)
    val cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITIES)
    val sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX)
    val keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS)
    val categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS)
 
    // 拼装过滤条件的字符串：
    var filterInfo = (if (startAge != null) Constants.PARAM_START_AGE + "=" + startAge + "|" else "") +
      (if (endAge != null) Constants.PARAM_END_AGE + "=" + endAge + "|" else "") +
      (if (professionals != null) Constants.PARAM_PROFESSIONALS + "=" + professionals + "|" else "") +
      (if (cities != null) Constants.PARAM_CITIES + "=" + cities + "|" else "") +
      (if (sex != null) Constants.PARAM_SEX + "=" + sex + "|" else "") +
      (if (keywords != null) Constants.PARAM_KEYWORDS + "=" + keywords + "|" else "") +
      (if (categoryIds != null) Constants.PARAM_CATEGORY_IDS + "=" + categoryIds else "")
 
    // 去除过滤条件字符串末尾的 "|"
    if (filterInfo.endsWith("\\|"))
      filterInfo = filterInfo.substring(0, filterInfo.length - 1)
 
    // 进行过滤操做（过滤自带遍历功能）
    sessionId2FullAggrInfoRDD.filter {
      case (sessionId, fullAggrInfo) =>
        var success = true
 
        // 若是 age 不在过滤条件范围以内，则当前 sessionId 对应的 fullAggrInfo 数据被过滤掉
        if (!ValidUtils.between(fullAggrInfo, Constants.FIELD_AGE, filterInfo, Constants.PARAM_START_AGE, Constants.PARAM_END_AGE)) { // 范围用 between
          success = false
        } else if (!ValidUtils.in(fullAggrInfo, Constants.FIELD_PROFESSIONAL, filterInfo, Constants.PARAM_PROFESSIONALS)) {
          success = false
        } else if (!ValidUtils.in(fullAggrInfo, Constants.FIELD_CITY, filterInfo, Constants.PARAM_CITIES)) {
          success = false
        } else if (!ValidUtils.equal(fullAggrInfo, Constants.FIELD_SEX, filterInfo, Constants.PARAM_SEX)) { // 二选一用 equal
          success = false
        } else if (!ValidUtils.in(fullAggrInfo, Constants.FIELD_SEARCH_KEYWORDS, filterInfo, Constants.PARAM_KEYWORDS)) { // 多选一用 in
          success = false
        } else if (!ValidUtils.in(fullAggrInfo, Constants.FIELD_CLICK_CATEGORY_IDS, filterInfo, Constants.PARAM_CATEGORY_IDS)) {
          success = false
        }
 
        // 自定义累加器，统计不一样范围的 访问时长 和 访问步长 的个数 以及 总的 session 个数
        if (success) {
          sessionStatisticAccumulator.add(Constants.SESSION_COUNT) // 总的 session 个数
 
          // 获取当前 sessionId 对应的 访问时长 和 访问步长
          val visitLength = StringUtils.getFieldFromConcatString(fullAggrInfo, "\\|", Constants.FIELD_VISIT_LENGTH).toLong
          val stepLength = StringUtils.getFieldFromConcatString(fullAggrInfo, "\\|", Constants.FIELD_STEP_LENGTH).toLong
 
          // 统计不一样范围的 访问时长 和 访问步长 的个数
          calculateVisitLength(visitLength, sessionStatisticAccumulator)
          calculateStepLength(stepLength, sessionStatisticAccumulator)
        }
 
        success
    }
  }
 
  def getSessionFullAggrInfo(sparkSession: SparkSession,
                             session2GroupActionRDD: RDD[(String, Iterable[UserVisitAction])]) = {
    // userId2PartAggrInfoRDD: RDD[(userId, partAggrInfo)]
    val userId2PartAggrInfoRDD = session2GroupActionRDD.map {
      // 使用模式匹配：当结果是 KV 对的时候尽可能使用 case 模式匹配，这样更清楚，更简洁直观
      case (sessionId, iterableAction) =>
        var userId = -1L
 
        var startTime: Date = null
        var endTime: Date = null
 
        var stepLength = 0 // 有多少个 action
 
        val searchKeywords = new StringBuffer("") // 搜索行为
        val clickCategories = new StringBuffer("") // 点击行为
 
        for (action <- iterableAction) {
          if (userId == -1) {
            userId = action.user_id
          }
 
          val actionTime = DateUtils.parseTime(action.action_time) // action_time = "2019-05-30 18:17:11" 是字符串类型
          if (startTime == null || startTime.after(actionTime)) { // startTime 在 actionTime 的后面   正常区间：[startTime, actionTime, endTime]
            startTime = actionTime
          }
 
          if (endTime == null || endTime.before(actionTime)) { // endTime 在 actionTime 的前面
            endTime = actionTime
          }
 
          val searchKeyword = action.search_keyword
          if (StringUtils.isNotEmpty(searchKeyword) && !searchKeywords.toString.contains(searchKeyword)) {
            searchKeywords.append(searchKeyword + ",")
          }
 
          val clickCategoryId = action.click_category_id
          if (clickCategoryId != -1 && !clickCategories.toString.contains(clickCategoryId)) {
            clickCategories.append(clickCategoryId + ",")
          }
 
          stepLength += 1
        }
 
        // searchKeywords.toString.substring(0, searchKeywords.toString.length - 1) // 等价于下面
        val searchKw = StringUtils.trimComma(searchKeywords.toString) // 去除最后一个逗号
        val clickCg = StringUtils.trimComma(clickCategories.toString) // 去除最后一个逗号
 
        val visitLength = (endTime.getTime - startTime.getTime) / 1000
 
        // 拼装聚合数据的字符串：
        // (31,sessionid=7291cc307f96432f8da9d926fd7d88e5|searchKeywords=洗面奶,小龙虾,机器学习,苹果,华为手机|clickCategoryIds=11,93,36,66,
        // 60|visitLength=3461|stepLength=43|startTime=2019-05-30 14:01:01)
        val partAggrInfo = Constants.FIELD_SESSION_ID + "=" + sessionId + "|" +
          Constants.FIELD_SEARCH_KEYWORDS + "=" + searchKw + "|" +
          Constants.FIELD_CLICK_CATEGORY_IDS + "=" + clickCg + "|" +
          Constants.FIELD_VISIT_LENGTH + "=" + visitLength + "|" +
          Constants.FIELD_STEP_LENGTH + "=" + stepLength + "|" +
          Constants.FIELD_START_TIME + "=" + DateUtils.formatTime(startTime) // 格式化时间为字符串类型
 
        (userId, partAggrInfo)
    }
 
    // user_visit_action 表联立 user_info 表，让咱们的统计数据中具备用户属性
    val sql = "select * from user_info"
 
    import sparkSession.implicits._
    // userId2InfoRDD: RDD[(userId, UserInfo)]
    val userId2InfoRDD = sparkSession.sql(sql).as[UserInfo].rdd.map(item => (item.user_id, item))
 
    val sessionId2FullAggrInfoRDD = userId2PartAggrInfoRDD.join(userId2InfoRDD).map {
      case (userId, (partAggrInfo, userInfo)) =>
        val age = userInfo.age
        val professional = userInfo.professional
        val sex = userInfo.sex
        val city = userInfo.city
 
        // 拼装最终的聚合数据字符串：
        val fullAggrInfo = partAggrInfo + "|" +
          Constants.FIELD_AGE + "=" + age + "|" +
          Constants.FIELD_PROFESSIONAL + "=" + professional + "|" +
          Constants.FIELD_SEX + "=" + sex + "|" +
          Constants.FIELD_CITY + "=" + city
 
        val seesionId = StringUtils.getFieldFromConcatString(partAggrInfo, "\\|", Constants.FIELD_SESSION_ID)
 
        (seesionId, fullAggrInfo)
    }
 
    sessionId2FullAggrInfoRDD
  }
 
  def getOriActionRDD(sparkSession: SparkSession, taskParam: JSONObject) = {
    // 先获取所用到的过滤条件：开始日期 和 结束日期
    val startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE)
    val endDate = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE)
    // 把全部的时间范围在 startDate 和 endDate 之间的数据查询出来
    val sql = "select * from user_visit_action where date>='" + startDate + "' and date<='" + endDate + "'"
 
    // 在对 DataFrame 和 Dataset 进行许多操做都须要这个包进行支持
    import sparkSession.implicits._
    sparkSession.sql(sql).as[UserVisitAction].rdd // DataFrame(Row类型) -> DataSet(样例类类型) -> rdd(样例类)
  }
}

自定义累加器 SessionStatisticAccumulator 代码以下：

package com.atguigu.session
 
import org.apache.spark.util.AccumulatorV2
 
import scala.collection.mutable
 
/**
  * 自定义累加器
  */
class SessionStatisticAccumulator extends AccumulatorV2[String, mutable.HashMap[String, Int]]() {
 
  // 自定义累加器：要求要在类的里面维护一个 mutable.HashMap 结构
  val countMap = new mutable.HashMap[String, Int]()
 
  // 判断累加器是否为空
  override def isZero: Boolean = {
    this.countMap.isEmpty
  }
 
  // 复制一个如出一辙的累加器
  override def copy(): AccumulatorV2[String, mutable.HashMap[String, Int]] = {
    val acc = new SessionStatisticAccumulator
    acc.countMap ++= this.countMap // 将两个 Map 拼接在一块儿
    acc
  }
 
  // 重置累加器
  override def reset(): Unit = {
    this.countMap.clear()
  }
 
  // 向累加器中添加 KV 对（K 存在，V 累加1，K 不存在，从新建立）
  override def add(k: String): Unit = {
    if (!this.countMap.contains(k)) {
      this.countMap += (k -> 0)
    }
 
    this.countMap.update(k, this.countMap(k) + 1)
  }
 
  // 两个累加器进行合并（先判断两个累加器是不是同一类型的，再将两个 Map 进行合并(是个小难点)）
  override def merge(other: AccumulatorV2[String, mutable.HashMap[String, Int]]): Unit = {
    other match {
      // (1 : 100).foldLeft(0) 等价于 (0 : (1 to 100))(_+_)  又等价于 { case (int1, int2) => int1 + int2 }
      // acc.countMap.foldLeft(this.countMap) 等价于 this.countMap : acc.countMap  又等价于 this.countMap 和 acc.countMap 的每个 KV 作操做
      case acc: SessionStatisticAccumulator => acc.countMap.foldLeft(this.countMap) {
        case (map, (k, v)) => map += (k -> (map.getOrElse(k, 0) + v))
      }
    }
  }
 
  override def value: mutable.HashMap[String, Int] = {
    this.countMap
  }
}

数据模型代码以下：

package com.atguigu.session
 
//***************** 输出表 *********************
 
/**
  * Session 聚合统计表
  *
  * @param taskid                     当前计算批次的 ID
  * @param session_count              全部 Session 的总和
  * @param visit_length_1s_3s_ratio   1-3s Session 访问时长占比
  * @param visit_length_4s_6s_ratio   4-6s Session 访问时长占比
  * @param visit_length_7s_9s_ratio   7-9s Session 访问时长占比
  * @param visit_length_10s_30s_ratio 10-30s Session 访问时长占比
  * @param visit_length_30s_60s_ratio 30-60s Session 访问时长占比
  * @param visit_length_1m_3m_ratio   1-3m Session 访问时长占比
  * @param visit_length_3m_10m_ratio  3-10m Session 访问时长占比
  * @param visit_length_10m_30m_ratio 10-30m Session 访问时长占比
  * @param visit_length_30m_ratio     30m Session 访问时长占比
  * @param step_length_1_3_ratio      1-3 步长占比
  * @param step_length_4_6_ratio      4-6 步长占比
  * @param step_length_7_9_ratio      7-9 步长占比
  * @param step_length_10_30_ratio    10-30 步长占比
  * @param step_length_30_60_ratio    30-60 步长占比
  * @param step_length_60_ratio       大于 60 步长占比
  */
case class SessionAggrStat(taskid: String,
                           session_count: Long,
                           visit_length_1s_3s_ratio: Double,
                           visit_length_4s_6s_ratio: Double,
                           visit_length_7s_9s_ratio: Double,
                           visit_length_10s_30s_ratio: Double,
                           visit_length_30s_60s_ratio: Double,
                           visit_length_1m_3m_ratio: Double,
                           visit_length_3m_10m_ratio: Double,
                           visit_length_10m_30m_ratio: Double,
                           visit_length_30m_ratio: Double,
                           step_length_1_3_ratio: Double,
                           step_length_4_6_ratio: Double,
                           step_length_7_9_ratio: Double,
                           step_length_10_30_ratio: Double,
                           step_length_30_60_ratio: Double,
                           step_length_60_ratio: Double)
 
/**
  * Session 随机抽取表
  *
  * @param taskid           当前计算批次的 ID
  * @param sessionid        抽取的 Session 的 ID
  * @param startTime        Session 的开始时间
  * @param searchKeywords   Session 的查询字段
  * @param clickCategoryIds Session 点击的类别 id 集合
  */
case class SessionRandomExtract(taskid: String,
                                sessionid: String,
                                startTime: String,
                                searchKeywords: String,
                                clickCategoryIds: String)
 
/**
  * Session 随机抽取详细表
  *
  * @param taskid           当前计算批次的 ID
  * @param userid           用户的 ID
  * @param sessionid        Session的 ID
  * @param pageid           某个页面的 ID
  * @param actionTime       点击行为的时间点
  * @param searchKeyword    用户搜索的关键词
  * @param clickCategoryId  某一个商品品类的 ID
  * @param clickProductId   某一个商品的 ID
  * @param orderCategoryIds 一次订单中全部品类的 ID 集合
  * @param orderProductIds  一次订单中全部商品的 ID 集合
  * @param payCategoryIds   一次支付中全部品类的 ID 集合
  * @param payProductIds    一次支付中全部商品的 ID 集合
  **/
case class SessionDetail(taskid: String,
                         userid: Long,
                         sessionid: String,
                         pageid: Long,
                         actionTime: String,
                         searchKeyword: String,
                         clickCategoryId: Long,
                         clickProductId: Long,
                         orderCategoryIds: String,
                         orderProductIds: String,
                         payCategoryIds: String,
                         payProductIds: String)
 
/**
  * 品类 Top10 表
  *
  * @param taskid
  * @param categoryid
  * @param clickCount
  * @param orderCount
  * @param payCount
  */
case class Top10Category(taskid: String,
                         categoryid: Long,
                         clickCount: Long,
                         orderCount: Long,
                         payCount: Long)
 
/**
  * Top10 Session
  *
  * @param taskid
  * @param categoryid
  * @param sessionid
  * @param clickCount
  */
case class Top10Session(taskid: String,
                        categoryid: Long,
                        sessionid: String,
                        clickCount: Long)

5.1.8 需求一实现思路整理

5.2 需求二：Session 随机抽取

5.2.1 需求解析

  在符合条件的 session 中，按照时间比例随机抽取 1000 个 session。
  这个按照时间比例是什么意思呢？随机抽取自己是很简单的，可是按照时间比例，就很复杂了。好比说，这一天总共有 1000 万的 session。那么我如今总共要从这 1000 万 session 中，随机抽取出来 1000 个 session。可是这个随机不是那么简单的。须要作到以下几点要求：首先，若是这一天的 12:00~13:00 的 session 数量是 100万，那么这个小时的 session 占比就是 1/10，那么这个小时中的 100 万的 session，咱们就要抽取 1/10 * 1000 = 100 个。即从这个小时的 100 万 session 中，随机抽取出 100 个 session。以此类推，其余小时的抽取也是这样作。
  这个功能的做用是说，可让使用者，可以对于符合条件的 session，按照时间比例均匀的随机采样出 1000 个 session，而后观察每一个 session 具体的点击流/行为，   好比先进入了首页、而后点击了食品品类、而后点击了雨润火腿肠商品、而后搜索了火腿肠罐头的关键词、接着对王中王火腿肠下了订单、最后对订单作了支付。
  之因此要作到按时间比例随机采用抽取，就是要作到，观察样本的公平性。
  抽取完毕以后，须要将 Session 的相关信息和详细信息保存到 MySQL 数据库中。

5.2.2 数据源解析

  本需求的数据源来自于需求一中获取的 Session 聚合数据（fullAggrInfo）。

5.2.3 数据结构解析

SessionRandomExtract 样例类

/**
  * Session 随机抽取表
  *
  * @param taskid           当前计算批次的 ID
  * @param sessionid        抽取的 Session 的 ID
  * @param startTime        Session 的开始时间
  * @param searchKeywords   Session 的查询字段
  * @param clickCategoryIds Session 点击的类别 id 集合
  */
case class SessionRandomExtract(taskid: String,
                                sessionid: String,
                                startTime: String,
                                searchKeywords: String,
                                clickCategoryIds: String)

5.2.4 需求实现简要流程

5.2.5 需求实现详细流程

5.2.6 MySQL 存储结构解析

MySQL 写入数据格式
session_detail

-- ----------------------------
--  Table structure for `session_detail`
-- ----------------------------
DROP TABLE IF EXISTS `session_detail`; 
CREATE TABLE `session_detail` (
  `taskid` varchar(255) DEFAULT NULL,
  `userid` int(11) DEFAULT NULL,
  `sessionid` varchar(255) DEFAULT NULL,
  `pageid` int(11) DEFAULT NULL,
  `actionTime` varchar(255) DEFAULT NULL,
  `searchKeyword` varchar(255) DEFAULT NULL,
  `clickCategoryId` int(11) DEFAULT NULL,
  `clickProductId` int(11) DEFAULT NULL,
  `orderCategoryIds` varchar(255) DEFAULT NULL,
  `orderProductIds` varchar(255) DEFAULT NULL,
  `payCategoryIds` varchar(255) DEFAULT NULL,
  `payProductIds` varchar(255) DEFAULT NULL, KEY `idx_task_id` (`taskid`),
  KEY `idx_session_id` (`sessionid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

session_random_extract

-- ----------------------------
--  Table structure for `session_random_extract`
-- ----------------------------
DROP TABLE IF EXISTS `session_random_extract`; 
CREATE TABLE `session_random_extract` (
  `taskid` varchar(255) DEFAULT NULL,
  `sessionid` varchar(255) DEFAULT NULL,
  `startTime` varchar(50) DEFAULT NULL,
  `searchKeywords` varchar(255) DEFAULT NULL,
  `clickCategoryIds` varchar(255) DEFAULT NULL, 
  KEY `idx_task_id` (`taskid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

5.2.7 代码解析

package com.atguigu.session
 
import java.util.{Date, Random, UUID}
 
import commons.conf.ConfigurationManager
import commons.constant.Constants
import commons.model.{UserInfo, UserVisitAction}
import commons.utils._
import net.sf.json.JSONObject
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SaveMode, SparkSession}
 
import scala.collection.mutable
import scala.collection.mutable.{ArrayBuffer, ListBuffer}
 
object SessionStat {
 
  def main(args: Array[String]): Unit = {
    // 获取过滤条件，【为了方便，直接从配置文件中获取，企业中会从一个调度平台获取】
    val jsonStr = ConfigurationManager.config.getString(Constants.TASK_PARAMS)
    // 获取过滤条件对应的 JsonObject 对象
    val taskParam = JSONObject.fromObject(jsonStr)
 
    // 建立全局惟一的主键，每次执行 main 函数都会生成一个独一无二的 taskUUID，来区分不一样任务，做为写入 MySQL 数据库中那张表的主键
    val taskUUID = UUID.randomUUID().toString
 
    // 建立 sparkConf
    val sparkConf = new SparkConf().setAppName("session").setMaster("local[*]")
 
    // 建立 sparkSession（包含 sparkContext）
    val sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()
 
    // ******************** 需求一：Session 各范围访问步长、访问时长占比统计 ********************
 
    // 获取原始的动做表数据（带有过滤条件）
    // actionRDD: RDD[UserVisitAction]
    val actionRDD = getOriActionRDD(sparkSession, taskParam)
 
    // 将用户行为信息转换为 K-V 结构，sessionId2ActionRDD: RDD[(sessionId, UserVisitAction)]
    val sessionId2ActionRDD = actionRDD.map(item => (item.session_id, item))
 
    // session2GroupActionRDD: RDD[(sessionId, Iterable[UserVisitAction])]
    val session2GroupActionRDD = sessionId2ActionRDD.groupByKey() // 把同一个 sessionId 的数据聚合到一块儿，获得斧子形数据
 
    // 将数据进行内存缓存
    session2GroupActionRDD.cache()
 
    // sessionId2FullAggrInfoRDD: RDD[(sessionId, fullAggrInfo)]
    val sessionId2FullAggrInfoRDD = getSessionFullAggrInfo(sparkSession, session2GroupActionRDD)
 
    // 建立自定义累加器对象
    val sessionStatisticAccumulator = new SessionStatisticAccumulator
 
    // 在 sparkSession 中注册自定义累加器，这样后面就能够用了
    sparkSession.sparkContext.register(sessionStatisticAccumulator)
 
    // 根据过滤条件对 sessionId2FullAggrInfoRDD 进行过滤操做，即过滤掉不符合条件的数据，并根据自定义累加器 统计不一样范围的 访问时长 和 访问步长 的 session 个数 以及 总的 session 个数
    val seeionId2FilterRDD = getSessionFilterRDD(taskParam, sessionId2FullAggrInfoRDD, sessionStatisticAccumulator)
 
    // 必须引入任意一个 action 的算子，才能启动
    seeionId2FilterRDD.foreach(println(_))
 
    // 计算各个 session 的占比
    getSessionRatio(sparkSession, taskUUID, sessionStatisticAccumulator.value)
 
    // ******************** 需求二：Session 随机抽取 ********************
 
    // sessionId2FullAggrInfoRDD: RDD[(sessionId, fullAggrInfo)]，注意：到这里一个 sessionId 对应一条数据，也就是一个 fullAggrInfo
    sessionRandomExtract(sparkSession, taskUUID, seeionId2FilterRDD)
 
  }
 
  // ******************** 需求二：Session 随机抽取 ********************
  /**
    * Session 随机抽取
    *
    * @param sparkSession
    * @param taskUUID
    * @param seeionId2FilterRDD
    */
  def sessionRandomExtract(sparkSession: SparkSession, taskUUID: String, seeionId2FilterRDD: RDD[(String, String)]): Unit = {
    // 因为是按照 时间 为 key 进行聚合，因此先将 seeionId2FilterRDD 的 key 转化为 时间
 
    // dateHour2FullAggrInfoRDD: RDD[(dateHour, fullAggrInfo)]
    val dateHour2FullAggrInfoRDD = seeionId2FilterRDD.map {
      case (sessionId, fullAggrInfo) =>
        // 先从 fullAggrInfo 中提取出来 startTime
        val startTime = StringUtils.getFieldFromConcatString(fullAggrInfo, "\\|", Constants.FIELD_START_TIME)
        // 获得的 startTime = "2019-05-30 18:17:11" 是字符串类型，须要转换成咱们须要的格式：yyyy-MM-dd_HH
        val dateHour = DateUtils.getDateHour(startTime)
 
        (dateHour, fullAggrInfo)
    }
 
    // hourCountMap: Map[(dateHour, count)]，示例：(yyyy-MM-dd_HH, 20)
    val hourCountMap = dateHour2FullAggrInfoRDD.countByKey()
 
    // dateHourCountMap: Map[data, Map[(hour, count)]]，示例：(yyyy-MM-dd, (HH, 20))
    val dateHourCountMap = new mutable.HashMap[String, mutable.HashMap[String, Long]]()
 
    for ((dateHour, count) <- hourCountMap) {
      val date = dateHour.split("_")(0) // yyyy-MM-dd_HH
      val hour = dateHour.split("_")(1) // HH
 
      dateHourCountMap.get(date) match { // Map[(hour, count)
        case None =>
          dateHourCountMap(date) = new mutable.HashMap[String, Long]() // 先建立 1 个空的 HashMap
          dateHourCountMap(date) += (hour -> count) // 再给 HashMap 赋值
        case Some(map) =>
          dateHourCountMap(date) += (hour -> count) // 直接给 HashMap 赋值
      }
    }
 
    // 解决问题一：
    //   一共有多少天：dateHourCountMap.size
    //   一天抽取多少条：1000 / dateHourCountMap.size
    val extractPerDay = 1000 / dateHourCountMap.size
 
    // 解决问题二：
    //   一共有多少个：session：dateHourCountMap(date).values.sum
    //   一个小时有多少个：session：dateHourCountMap(date)(hour)
 
    val dateHourExtractIndexListMap = new mutable.HashMap[String, mutable.HashMap[String, ListBuffer[Int]]]()
 
    // dateHourCountMap: Map[data, Map[(hour, count)]]，示例：(yyyy-MM-dd, (HH, 20))
    // hourCountMap: Map[(hour, count)]，示例：(HH, 20) ，注意：这里面的 hourCountMap 含义发生变化了，要跟上面的最开始的 hourCountMap 区别开来
    for ((date, hourCountMap) <- dateHourCountMap) {
      // 一天共有多少个 session
      val dataCount = hourCountMap.values.sum
 
      dateHourExtractIndexListMap.get(date) match {
        case None =>
          dateHourExtractIndexListMap(date) = new mutable.HashMap[String, mutable.ListBuffer[Int]]()
          generateRandomIndexList(extractPerDay, dataCount, hourCountMap, dateHourExtractIndexListMap(date))
        case Some(map) =>
          generateRandomIndexList(extractPerDay, dataCount, hourCountMap, dateHourExtractIndexListMap(date))
      }
    }
 
    // 到此为止，咱们得到了每一个小时要抽取的 session 的 index
    // 以后在算子中使用 dateHourExtractIndexListMap 这个 Map，因为这个 Map 可能会很大，因此涉及到 广播大变量 的问题
 
    // 广播大变量，提高任务 task 的性能
    val dateHourExtractIndexListMapBroadcastVar = sparkSession.sparkContext.broadcast(dateHourExtractIndexListMap)
 
    // dateHour2FullAggrInfoRDD: RDD[(dateHour, fullAggrInfo)]
    // dateHour2GroupRDD: RDD[(dateHour, Iterable[fullAggrInfo])]
    val dateHour2GroupRDD = dateHour2FullAggrInfoRDD.groupByKey()
 
    // extractSessionRDD: RDD[SessionRandomExtract]
    val extractSessionRDD = dateHour2GroupRDD.flatMap {
      case (dateHour, iterableFullAggrInfo) =>
        val date = dateHour.split("_")(0)
        val hour = dateHour.split("_")(1)
 
        val extractIndexList = dateHourExtractIndexListMapBroadcastVar.value.get(date).get(hour)
 
        // 建立一个容器存储抽取的 session
        val extractSessionArrayBuffer = new ArrayBuffer[SessionRandomExtract]()
        var index = 0
 
        for (fullAggrInfo <- iterableFullAggrInfo) {
          if (extractIndexList.contains(index)) {
            // 提取数据，封装成所须要的样例类，并追加进 ArrayBuffer 中
            val sessionId = StringUtils.getFieldFromConcatString(fullAggrInfo, "\\|", Constants.FIELD_SESSION_ID)
            val startTime = StringUtils.getFieldFromConcatString(fullAggrInfo, "\\|", Constants.FIELD_START_TIME)
            val searchKeywords = StringUtils.getFieldFromConcatString(fullAggrInfo, "\\|", Constants.FIELD_SEARCH_KEYWORDS)
            val clickCategoryIds = StringUtils.getFieldFromConcatString(fullAggrInfo, "\\|", Constants.FIELD_CLICK_CATEGORY_IDS)
 
            val sessionRandomExtract = SessionRandomExtract(taskUUID, sessionId, startTime, searchKeywords, clickCategoryIds)
 
            extractSessionArrayBuffer += sessionRandomExtract
          }
          index += 1
        }
 
        extractSessionArrayBuffer
    }
 
    // 将抽取后的数据保存到 MySQL
    import sparkSession.implicits._
    extractSessionRDD.toDF().write
      .format("jdbc")
      .option("url", ConfigurationManager.config.getString(Constants.JDBC_URL))
      .option("dbtable", "session_random_extract")
      .option("user", ConfigurationManager.config.getString(Constants.JDBC_USER))
      .option("password", ConfigurationManager.config.getString(Constants.JDBC_PASSWORD))
      .mode(SaveMode.Append)
      .save()
  }
 
  /**
    * 根据每一个小时应该抽取的数量，来产生随机值
    *
    * @param extractPerDay 一天抽取的 seesion 个数
    * @param dataCount     当天全部的 seesion 总数
    * @param hourCountMap  每一个小时的session总数
    * @param hourListMap   主要用来存放生成的随机值
    */
  def generateRandomIndexList(extractPerDay: Long,
                              dataCount: Long,
                              hourCountMap: mutable.HashMap[String, Long],
                              hourListMap: mutable.HashMap[String, mutable.ListBuffer[Int]]): Unit = {
    // 先遍历 hourCountMap，hourCountMap: Map[(hour, count)]，示例：(HH, 20) ，注意：这里面的 hourCountMap 含义发生变化了，要跟上面的最开始的 hourCountMap 区别开来
    for ((hour, count) <- hourCountMap) {
      // 计算一个小时抽取多少个 session
      var hourExtractCount = ((count / dataCount.toDouble) * extractPerDay).toInt
      // 避免一个小时要抽取的数量超过这个小时的总数
      if (hourExtractCount > count) {
        hourExtractCount = count.toInt
      }
 
      val random = new Random()
 
      hourListMap.get(hour) match {
        case None =>
          hourListMap(hour) = new mutable.ListBuffer[Int] // 没有 List，须要新建一个 List
          for (i <- 0 until hourExtractCount) {
            var index = random.nextInt(count.toInt) // 生成 index
            while (hourListMap(hour).contains(index)) { // 若是 index 已存在
              index = random.nextInt(count.toInt) // 则从新生成 index
            }
 
            // 将生成的 index 放入到 hourListMap 中
            hourListMap(hour).append(index)
          }
 
        case Some(list) =>
          for (i <- 0 until hourExtractCount) {
            var index = random.nextInt(count.toInt) // 生成 index
            while (hourListMap(hour).contains(index)) { // 若是 index 已存在
              index = random.nextInt(count.toInt) // 则从新生成 index
            }
 
            // 将生成的 index 放入到 hourListMap 中
            hourListMap(hour).append(index)
          }
      }
    }
 
 
  }
 
  // ******************** 需求一：Session 各范围访问步长、访问时长占比统计 ********************
  // ......
}

5.2.8 需求二实现思路整理

5.3 需求三：Top10 热门品类统计

5.3.1 需求解析

  在符合条件的 session 中，获取点击、下单和支付数量排名前 10 的品类。
  数据中的每一个 session 可能都会对一些品类的商品进行点击、下单和支付等等行为，那么如今就须要获取这些 session 点击、下单和支付数量排名前 10 的最热门的品类。也就是说，要计算出全部这些 session 对各个品类的点击、下单和支付的次数， 而后按照这三个属性进行排序，获取前 10 个品类。
  这个功能很重要，可让咱们明白，符合条件的用户，他最感兴趣的商品是什么种类。这个可让公司里的人，清晰地了解到不一样层次、不一样类型的用户的心理和喜爱。
  计算完成以后，将数据保存到 MySQL 数据库中。

5.3.2 数据源解析

sessionId2ActionRDD: RDD[(sessionId, UserVisitAction)]
seeionId2FilterRDD: RDD[(sessionId, fullAggrInfo)]
 
seeionId2ActionFilterRDD: RDD[(sessionId, UserVisitAction)]

5.3.3 数据结构解析

/**
  * 品类 Top10 表
  *
  * @param taskid
  * @param categoryid
  * @param clickCount
  * @param orderCount
  * @param payCount
  */
case class Top10Category(taskid: String,
                         categoryid: Long,
                         clickCount: Long,
                         orderCount: Long,
                         payCount: Long)

5.3.4 需求实现简要流程

略

5.3.5 需求实现详细流程

5.3.6 MySQL 存储结构解析

-- ----------------------------
--  Table structure for `top10_category`
-- ----------------------------
DROP TABLE IF EXISTS `top10_category`;
CREATE TABLE `top10_category` (
  `taskid` varchar(255) DEFAULT NULL,
  `categoryid` int(11) DEFAULT NULL,
  `clickCount` int(11) DEFAULT NULL,
  `orderCount` int(11) DEFAULT NULL,
  `payCount` int(11) DEFAULT NULL,
  KEY `idx_task_id` (`taskid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

5.3.7 代码解析

package com.atguigu.session
 
import java.util.{Date, Random, UUID}
 
import commons.conf.ConfigurationManager
import commons.constant.Constants
import commons.model.{UserInfo, UserVisitAction}
import commons.utils._
import net.sf.json.JSONObject
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SaveMode, SparkSession}
 
import scala.collection.mutable
import scala.collection.mutable.{ArrayBuffer, ListBuffer}
 
object SessionStat {
 
  def main(args: Array[String]): Unit = {
    // 获取过滤条件，【为了方便，直接从配置文件中获取，企业中会从一个调度平台获取】
    val jsonStr = ConfigurationManager.config.getString(Constants.TASK_PARAMS)
    // 获取过滤条件对应的 JsonObject 对象
    val taskParam = JSONObject.fromObject(jsonStr)
 
    // 建立全局惟一的主键，每次执行 main 函数都会生成一个独一无二的 taskUUID，来区分不一样任务，做为写入 MySQL 数据库中那张表的主键
    val taskUUID = UUID.randomUUID().toString
 
    // 建立 sparkConf
    val sparkConf = new SparkConf().setAppName("session").setMaster("local[*]")
 
    // 建立 sparkSession（包含 sparkContext）
    val sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()
 
    // ******************** 需求一：Session 各范围访问步长、访问时长占比统计 ********************
 
    // 获取原始的动做表数据（带有过滤条件）
    // actionRDD: RDD[UserVisitAction]
    val actionRDD = getOriActionRDD(sparkSession, taskParam)
 
    // 将用户行为信息转换为 K-V 结构
    // sessionId2ActionRDD: RDD[(sessionId, UserVisitAction)]
    val sessionId2ActionRDD = actionRDD.map(item => (item.session_id, item))
 
    // session2GroupActionRDD: RDD[(sessionId, Iterable[UserVisitAction])]
    val session2GroupActionRDD = sessionId2ActionRDD.groupByKey() // 把同一个 sessionId 的数据聚合到一块儿，获得斧子形数据
 
    // 将数据进行内存缓存
    session2GroupActionRDD.cache()
 
    // sessionId2FullAggrInfoRDD: RDD[(sessionId, fullAggrInfo)]
    val sessionId2FullAggrInfoRDD = getSessionFullAggrInfo(sparkSession, session2GroupActionRDD)
 
    // 建立自定义累加器对象
    val sessionStatisticAccumulator = new SessionStatisticAccumulator
 
    // 在 sparkSession 中注册自定义累加器，这样后面就能够用了
    sparkSession.sparkContext.register(sessionStatisticAccumulator)
 
    // 根据过滤条件对 sessionId2FullAggrInfoRDD 进行过滤操做，即过滤掉不符合条件的数据，并根据自定义累加器 统计不一样范围的 访问时长 和 访问步长 的 session 个数 以及 总的 session 个数
    // seeionId2FilterRDD: RDD[(sessionId, fullAggrInfo)]
    val seeionId2FilterRDD = getSessionFilterRDD(taskParam, sessionId2FullAggrInfoRDD, sessionStatisticAccumulator)
 
    // 必须引入任意一个 action 的算子，才能启动
    seeionId2FilterRDD.foreach(println(_))
 
    // 计算各个 session 的占比
    getSessionRatio(sparkSession, taskUUID, sessionStatisticAccumulator.value)
 
    // ******************** 需求二：Session 随机抽取 ********************
 
    // sessionId2FullAggrInfoRDD: RDD[(sessionId, fullAggrInfo)]，注意：到这里一个 sessionId 对应一条数据，也就是一个 fullAggrInfo
    sessionRandomExtract(sparkSession, taskUUID, seeionId2FilterRDD)
 
    // ******************** 需求三：Top10 热门品类统计 ********************
 
    // sessionId2ActionRDD: RDD[(sessionId, UserVisitAction)]
    // seeionId2FilterRDD: RDD[(sessionId, fullAggrInfo)]
 
    // join 默认是内链接，即不符合条件的不显示（即被过滤掉）
 
    // 获取全部符合过滤条件的原始的 UserVisitAction 数据
    val seeionId2ActionFilterRDD = sessionId2ActionRDD.join(seeionId2FilterRDD).map {
      case (sessionId, (userVisitAction, fullAggrInfo)) =>
        (sessionId, userVisitAction)
    }
 
    val top10CategoryArray = top10PopularCategories(sparkSession, taskUUID, seeionId2ActionFilterRDD)
  }
 
  /**
    * Top10 热门品类统计
    *
    * @param sparkSession
    * @param taskUUID
    * @param seeionId2ActionFilterRDD 全部符合过滤条件的原始的 UserVisitAction 数据
    */
  def top10PopularCategories(sparkSession: SparkSession, taskUUID: String, seeionId2ActionFilterRDD: RDD[(String, UserVisitAction)]) = {
    // 第一步：获取全部发生过点击、下单、付款的 categoryId，注意：其中被点击的 categoryId 只有一个，被下单和被付款的 categoryId 有多个，categoryId 之间使用逗号隔开的
    var cid2CidRDD = seeionId2ActionFilterRDD.flatMap {
      case (sessionId, userVisitAction) =>
        val categoryIdBuffer = new ArrayBuffer[(Long, Long)]()
 
        // 提取出数据填充 ArrayBuffer
        if (userVisitAction.click_category_id != -1) { // 点击行为
          categoryIdBuffer += ((userVisitAction.click_category_id, userVisitAction.click_category_id)) // 只有第一个 key 有用，第二个 value 任何值均可以，可是不能够没有
        } else if (userVisitAction.order_category_ids != null) { // 下单行为
          for (order_category_id <- userVisitAction.order_category_ids.split(",")) {
            categoryIdBuffer += ((order_category_id.toLong, order_category_id.toLong))
          }
        } else if (userVisitAction.pay_category_ids != null) { // 付款行为
          for (pay_category_id <- userVisitAction.pay_category_ids.split(",")) {
            categoryIdBuffer += ((pay_category_id.toLong, pay_category_id.toLong))
          }
        }
 
        categoryIdBuffer
    }
 
    // 第二步：进行去重操做
    cid2CidRDD = cid2CidRDD.distinct()
 
    // 第三步：统计各品类 被点击的次数、被下单的次数、被付款的次数
    val cid2ClickCountRDD = getClickCount(seeionId2ActionFilterRDD)
    val cid2OrderCountRDD = getOrderCount(seeionId2ActionFilterRDD)
    val cid2PayCountRDD = getPayCount(seeionId2ActionFilterRDD)
 
    // 第四步：获取各个 categoryId 的点击次数、下单次数、付款次数，并进行拼装
    // cid2FullCountRDD: RDD[(cid, aggrCountInfo)]
    // (81,categoryId=81|clickCount=68|orderCount=64|payCount=72)
    val cid2FullCountRDD = getFullCount(cid2CidRDD, cid2ClickCountRDD, cid2OrderCountRDD, cid2PayCountRDD)
 
    // 第五步：根据点击次数、下单次数、付款次数依次排序，会用到 【二次排序】，实现自定义的二次排序的 key
 
    // 第六步：封装 SortKey
    val sortKey2FullCountRDD = cid2FullCountRDD.map {
      case (cid, fullCountInfo) =>
        val clickCount = StringUtils.getFieldFromConcatString(fullCountInfo, "\\|", Constants.FIELD_CLICK_COUNT).toLong
        val orderCount = StringUtils.getFieldFromConcatString(fullCountInfo, "\\|", Constants.FIELD_ORDER_COUNT).toLong
        val payCount = StringUtils.getFieldFromConcatString(fullCountInfo, "\\|", Constants.FIELD_PAY_COUNT).toLong
 
        val sortKey = SortKey(clickCount, orderCount, payCount)
        (sortKey, fullCountInfo)
    }
 
    // 第七步：降序排序，取出 top10 热门品类
    val top10CategoryArray = sortKey2FullCountRDD.sortByKey(false).take(10)
 
    // 第八步：将 Array 结构转化为 RDD，封装 Top10Category
    val top10CategoryRDD = sparkSession.sparkContext.makeRDD(top10CategoryArray).map {
      case (sortKey, fullCountInfo) =>
        val categoryid = StringUtils.getFieldFromConcatString(fullCountInfo, "\\|", Constants.FIELD_CATEGORY_ID).toLong
        val clickCount = sortKey.clickCount
        val orderCount = sortKey.orderCount
        val payCount = sortKey.payCount
 
        Top10Category(taskUUID, categoryid, clickCount, orderCount, payCount)
    }
 
    // 第九步：写入 MySQL 数据库
    import sparkSession.implicits._
    top10CategoryRDD.toDF().write
      .format("jdbc")
      .option("url", ConfigurationManager.config.getString(Constants.JDBC_URL))
      .option("dbtable", "top10_category")
      .option("user", ConfigurationManager.config.getString(Constants.JDBC_USER))
      .option("password", ConfigurationManager.config.getString(Constants.JDBC_PASSWORD))
      .mode(SaveMode.Append)
      .save()
 
    top10CategoryArray
  }
 
  /**
    *
    * @param cid2CidRDD
    * @param cid2ClickCountRDD
    * @param cid2OrderCountRDD
    * @param cid2PayCountRDD
    * @return
    */
  def getFullCount(cid2CidRDD: RDD[(Long, Long)],
                   cid2ClickCountRDD: RDD[(Long, Long)],
                   cid2OrderCountRDD: RDD[(Long, Long)],
                   cid2PayCountRDD: RDD[(Long, Long)]) = {
    // 左外链接：不符合添加显示为空（null）
 
    // 4.1 全部品类id 和 被点击的品类 作左外链接
    val cid2ClickInfoRDD = cid2CidRDD.leftOuterJoin(cid2ClickCountRDD).map {
      case (cid, (categoryId, option)) =>
        val clickCount = if (option.isDefined) option.get else 0
        val aggrCountInfo = Constants.FIELD_CATEGORY_ID + "=" + cid + "|" + Constants.FIELD_CLICK_COUNT + "=" + clickCount
 
        (cid, aggrCountInfo)
    }
    // 4.2 4.1 的结果 和 被下单的品类 作左外链接
    val cid2OrderInfoRDD = cid2ClickInfoRDD.leftOuterJoin(cid2OrderCountRDD).map {
      case (cid, (clickInfo, option)) =>
        val orderCount = if (option.isDefined) option.get else 0
        val aggrCountInfo = clickInfo + "|" + Constants.FIELD_ORDER_COUNT + "=" + orderCount
 
        (cid, aggrCountInfo)
    }
    // 4.3 4.2 的结果 和 被付款的品类 作左外链接
    val cid2PayInfoRDD = cid2OrderInfoRDD.leftOuterJoin(cid2PayCountRDD).map {
      case (cid, (orderInfo, option)) =>
        val payCount = if (option.isDefined) option.get else 0
        val aggrCountInfo = orderInfo + "|" + Constants.FIELD_PAY_COUNT + "=" + payCount
 
        (cid, aggrCountInfo)
    }
 
    cid2PayInfoRDD
  }
 
  /**
    * 统计各品类被点击的次数
    *
    * @param seeionId2ActionFilterRDD
    */
  def getClickCount(seeionId2ActionFilterRDD: RDD[(String, UserVisitAction)]) = {
    // 方式一：把发生过点击的 action 过滤出来
    val clickActionFilterRDD = seeionId2ActionFilterRDD.filter {
      case (sessionId, userVisitAction) =>
        userVisitAction.click_category_id != 1L
    }
    // 方式二：把发生点击的 action 过滤出来，两者等价
    // val clickActionFilterRDD2 = seeionId2ActionFilterRDD.filter(item => item._2.click_category_id != -1L)
 
    // 获取每种类别的点击次数
    val clickNumRDD = clickActionFilterRDD.map {
      case (sessionId, userVisitAction) =>
        (userVisitAction.click_category_id, 1L)
    }
    // 计算各个品类的点击次数
    clickNumRDD.reduceByKey(_ + _)
  }
 
  /**
    * 统计各品类被下单的次数
    *
    * @param seeionId2ActionFilterRDD
    */
  def getOrderCount(seeionId2ActionFilterRDD: RDD[(String, UserVisitAction)]) = {
    // 把发生过下单的 action 过滤出来
    val orderActionFilterRDD = seeionId2ActionFilterRDD.filter {
      case (sessionId, userVisitAction) =>
        userVisitAction.order_category_ids != null
    }
    // 获取每种类别的下单次数
    val orderNumRDD = orderActionFilterRDD.flatMap {
      case (sessionId, userVisitAction) =>
        userVisitAction.order_category_ids.split(",").map(item => (item.toLong, 1L))
    }
    // 计算各个品类的下单次数
    orderNumRDD.reduceByKey(_ + _)
  }
 
  /**
    * 统计各品类被付款的次数
    *
    * @param seeionId2ActionFilterRDD
    */
  def getPayCount(seeionId2ActionFilterRDD: RDD[(String, UserVisitAction)]) = {
    // 把发生过付款的 action 过滤出来
    val payActionFilterRDD = seeionId2ActionFilterRDD.filter {
      case (sessionId, userVisitAction) =>
        userVisitAction.pay_category_ids != null
    }
    // 获取每种类别的支付次数
    val payNumRDD = payActionFilterRDD.flatMap {
      case (sessionId, userVisitAction) =>
        userVisitAction.pay_category_ids.split(",").map(item => (item.toLong, 1L))
    }
    // 计算各个品类的支付次数
    payNumRDD.reduceByKey(_ + _)
  }
 
  // ******************** 需求二：Session 随机抽取 ********************
 
  // ******************** 需求一：Session 各范围访问步长、访问时长占比统计 ********************
 
}

5.3.8 需求三实现思路整理

略

5.4 需求四：Top10 热门品类的 Top10 活跃 Session 统计

5.4.1 需求解析

  对于排名前 10 的品类，分别获取其点击次数排名前 10 的 session。
  这个就是说，对于 top10 的品类，每个都要获取对它点击次数排名前 10 的 session。
  这个功能，可让咱们看到，对某个用户群体最感兴趣的品类，各个品类最感兴趣最典型的用户的 session 的行为。
  计算完成以后，将数据保存到 MySQL 数据库中。

5.4.2 数据源解析

seeionId2ActionFilterRDD: RDD[(sessionId, UserVisitAction)]
top10CategoryArray: Array[(sortKey, fullCountInfo)]

5.4.3 数据结构解析

/**
  * Top10 Session
  *
  * @param taskid
  * @param categoryid
  * @param sessionid
  * @param clickCount
  */
case class Top10Session(taskid: String,
                        categoryid: Long,
                        sessionid: String,
                        clickCount: Long)
 
/**
  * Session 随机抽取详细表
  *
  * @param taskid           当前计算批次的 ID
  * @param userid           用户的 ID
  * @param sessionid        Session 的 ID
  * @param pageid           某个页面的 ID
  * @param actionTime       点击行为的时间点
  * @param searchKeyword    用户搜索的关键词
  * @param clickCategoryId  某一个商品品类的 ID
  * @param clickProductId   某一个商品的 ID
  * @param orderCategoryIds 一次订单中全部品类的 ID 集合
  * @param orderProductIds  一次订单中全部商品的 ID 集合
  * @param payCategoryIds   一次支付中全部品类的 ID 集合
  * @param payProductIds    一次支付中全部商品的 ID 集合
  **/
case class SessionDetail(taskid: String,
                         userid: Long,
                         sessionid: String,
                         pageid: Long,
                         actionTime: String,
                         searchKeyword: String,
                         clickCategoryId: Long,
                         clickProductId: Long,
                         orderCategoryIds: String,
                         orderProductIds: String,
                         payCategoryIds: String,
                         payProductIds: String)

5.4.4 需求实现简要流程

5.4.5 需求实现详细流程

略

5.4.6 MySQL 存储结构解析

-- ----------------------------
--  Table structure for `top10_session`
-- ----------------------------
DROP TABLE IF EXISTS `top10_session`;
CREATE TABLE `top10_session` (
  `taskid` varchar(255) DEFAULT NULL,
  `categoryid` int(11) DEFAULT NULL,
  `sessionid` varchar(255) DEFAULT NULL,
  `clickCount` int(11) DEFAULT NULL, KEY `idx_task_id` (`taskid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

5.4.7 代码解析

package com.atguigu.session
 
import java.util.{Date, Random, UUID}
 
import commons.conf.ConfigurationManager
import commons.constant.Constants
import commons.model.{UserInfo, UserVisitAction}
import commons.utils._
import net.sf.json.JSONObject
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SaveMode, SparkSession}
 
import scala.collection.mutable
import scala.collection.mutable.{ArrayBuffer, ListBuffer}
 
object SessionStat {
 
  def main(args: Array[String]): Unit = {
    // 获取过滤条件，【为了方便，直接从配置文件中获取，企业中会从一个调度平台获取】
    val jsonStr = ConfigurationManager.config.getString(Constants.TASK_PARAMS)
    // 获取过滤条件对应的 JsonObject 对象
    val taskParam = JSONObject.fromObject(jsonStr)
 
    // 建立全局惟一的主键，每次执行 main 函数都会生成一个独一无二的 taskUUID，来区分不一样任务，做为写入 MySQL 数据库中那张表的主键
    val taskUUID = UUID.randomUUID().toString
 
    // 建立 sparkConf
    val sparkConf = new SparkConf().setAppName("session").setMaster("local[*]")
 
    // 建立 sparkSession（包含 sparkContext）
    val sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()
 
    // ******************** 需求一：Session 各范围访问步长、访问时长占比统计 ********************
 
    // 获取原始的动做表数据（带有过滤条件）
    // actionRDD: RDD[UserVisitAction]
    val actionRDD = getOriActionRDD(sparkSession, taskParam)
 
    // 将用户行为信息转换为 K-V 结构
    // sessionId2ActionRDD: RDD[(sessionId, UserVisitAction)]
    val sessionId2ActionRDD = actionRDD.map(item => (item.session_id, item))
 
    // session2GroupActionRDD: RDD[(sessionId, Iterable[UserVisitAction])]
    val session2GroupActionRDD = sessionId2ActionRDD.groupByKey() // 把同一个 sessionId 的数据聚合到一块儿，获得斧子形数据
 
    // 将数据进行内存缓存
    session2GroupActionRDD.cache()
 
    // sessionId2FullAggrInfoRDD: RDD[(sessionId, fullAggrInfo)]
    val sessionId2FullAggrInfoRDD = getSessionFullAggrInfo(sparkSession, session2GroupActionRDD)
 
    // 建立自定义累加器对象
    val sessionStatisticAccumulator = new SessionStatisticAccumulator
 
    // 在 sparkSession 中注册自定义累加器，这样后面就能够用了
    sparkSession.sparkContext.register(sessionStatisticAccumulator)
 
    // 根据过滤条件对 sessionId2FullAggrInfoRDD 进行过滤操做，即过滤掉不符合条件的数据，并根据自定义累加器 统计不一样范围的 访问时长 和 访问步长 的 session 个数 以及 总的 session 个数
    // seeionId2FilterRDD: RDD[(sessionId, fullAggrInfo)]
    val seeionId2FilterRDD = getSessionFilterRDD(taskParam, sessionId2FullAggrInfoRDD, sessionStatisticAccumulator)
 
    // 必须引入任意一个 action 的算子，才能启动
    seeionId2FilterRDD.foreach(println(_))
 
    // 计算各个 session 的占比
    getSessionRatio(sparkSession, taskUUID, sessionStatisticAccumulator.value)
 
    // ******************** 需求二：Session 随机抽取 ********************
 
    // sessionId2FullAggrInfoRDD: RDD[(sessionId, fullAggrInfo)]，注意：到这里一个 sessionId 对应一条数据，也就是一个 fullAggrInfo
    sessionRandomExtract(sparkSession, taskUUID, seeionId2FilterRDD)
 
    // ******************** 需求三：Top10 热门品类统计 ********************
 
    // sessionId2ActionRDD: RDD[(sessionId, UserVisitAction)]
    // seeionId2FilterRDD: RDD[(sessionId, fullAggrInfo)]
 
    // join 默认是内链接，即不符合条件的不显示（即被过滤掉）
 
    // 获取全部符合过滤条件的原始的 UserVisitAction 数据
    // seeionId2ActionFilterRDD: RDD[(sessionId, UserVisitAction)]
    val seeionId2ActionFilterRDD = sessionId2ActionRDD.join(seeionId2FilterRDD).map {
      case (sessionId, (userVisitAction, fullAggrInfo)) =>
        (sessionId, userVisitAction)
    }
 
    val top10CategoryArray = top10PopularCategories(sparkSession, taskUUID, seeionId2ActionFilterRDD)
 
    // ******************** 需求四：Top10 热门品类的 Top10 活跃 Session 统计 ********************
 
    // seeionId2ActionFilterRDD: RDD[(sessionId, UserVisitAction)]
    // top10CategoryArray: Array[(sortKey, fullCountInfo)]
 
    top10ActiveSession(sparkSession, taskUUID, seeionId2ActionFilterRDD, top10CategoryArray)
 
  }
 
  // ******************** 需求四：Top10 热门品类的 Top10 活跃 Session 统计 ********************
 
  /**
    * Top10 热门品类的 Top10 活跃 Session 统计
    *
    * @param sparkSession
    * @param taskUUID
    * @param seeionId2ActionFilterRDD
    * @param top10CategoryArray
    */
  def top10ActiveSession(sparkSession: SparkSession,
                         taskUUID: String,
                         seeionId2ActionFilterRDD: RDD[(String, UserVisitAction)],
                         top10CategoryArray: Array[(SortKey, String)]): Unit = {
    // 第一步：获取全部点击过 Top10 热门品类的 UserVisitAction
    // 第一种方法：Join 方法，该方式须要引发 Shuffle，比较麻烦
 
    /*
    // 将 top10CategoryArray 转化为 RDD，而后将其 key sortKey 转化为 cid
    val cid2FullCountInfoRDD = sparkSession.sparkContext.makeRDD(top10CategoryArray).map {
      case (sortKey, fullCountInfo) =>
        // 取出 categoryId
        val cid = StringUtils.getFieldFromConcatString(fullCountInfo, "\\|", Constants.FIELD_CATEGORY_ID).toLong
        // 返回所需的 RDD
        (cid, fullCountInfo)
    }
 
    // 将 seeionId2ActionFilterRDD 的 key sessionId 转化为 cid，对其使用 map 操做便可
    val cid2ActionRDD = seeionId2ActionFilterRDD.map {
      case (sessionId, userVisitAction) =>
        val cid = userVisitAction.click_category_id
        (cid, userVisitAction)
    }
 
    // joinn 操做（即内链接）：两边都有的才留下，不然过滤掉
    cid2FullCountInfoRDD.join(cid2ActionRDD).map {
      case (cid, (fullCountInfo, userVisitAction)) =>
        val sid = userVisitAction.session_id
        (sid, userVisitAction)
    }*/
 
    // 第二种方法：使用 filter
    // cidArray: Array[Long] 包含了 Top10 热门品类的 id
    val cidArray = top10CategoryArray.map {
      case (sortKey, fullCountInfo) =>
        val cid = StringUtils.getFieldFromConcatString(fullCountInfo, "\\|", Constants.FIELD_CATEGORY_ID).toLong
        cid
    }
 
    // 全部符合过滤条件的，而且点击过 Top10 热门品类的 UserVisitAction
    val seeionId2ActionRDD = seeionId2ActionFilterRDD.filter {
      case (sessionId, userVisitAction) =>
        cidArray.contains(userVisitAction.click_category_id)
    }
 
    // 第二步：先对 全部符合过滤条件的，而且点击过 Top10 热门品类的 UserVisitAction 按照 sessionId 进行聚合
    val seeionId2GroupRDD = seeionId2ActionRDD.groupByKey()
 
 
    // 第三步：统计 每个 sessionId 对于点击过的每个品类的点击次数
    // cid2SessionCountRDD: RDD[(cid, sessionN=sessionCount)]
    val cid2SessionCountRDD = seeionId2GroupRDD.flatMap {
      case (sessionId, iterableUserVisitAction) =>
        // 建立 Map，用于保存当前每个 sessionId 对于点击过的每个品类的点击次数
        val categoryCountMap = new mutable.HashMap[Long, Long]()
 
        for (userVisitAction <- iterableUserVisitAction) {
          val cid = userVisitAction.click_category_id
          if (!categoryCountMap.contains(cid))
            categoryCountMap += (cid -> 0)
 
          categoryCountMap.update(cid, categoryCountMap(cid) + 1)
        }
 
        // 该 Map 记录了一个 session 对于它全部点击过的品类的点击次数
        // categoryCountMap
 
        for ((cid, sessionCount) <- categoryCountMap)
          yield (cid, sessionId + "=" + sessionCount)
    }
 
    // 第四步：对 cid2SessionCountRDD 进行聚合
    // cid2GroupRDD: RDD[(cid, Iterable[sessionN=sessionCount]))]
    // cid2GroupRDD 的每一条数据都是一个 cid 和它对应的全部点击过它的 sessionId 对它的点击次数
    val cid2GroupRDD = cid2SessionCountRDD.groupByKey()
 
    // 第五步：取出 top10SessionRDD: RDD[Top10Session]
    val top10SessionRDD = cid2GroupRDD.flatMap {
      case (cid, iterablesSessionCount) =>
        val sortList = iterablesSessionCount.toList.sortWith((item1, item2) => { // true: item1 放在前面
          item1.split("=")(1).toLong > item2.split("=")(1).toLong // item1: sessionCount 字符串类型 sessionIdN=count
        }).take(10)
 
        // 封装数据，准备写入 MySQL 数据库
        val top10Session = sortList.map {
          case item => {
            val categoryid = cid
            val sessionid = item.split("=")(0)
            val clickCount = item.split("=")(1).toLong
 
            Top10Session(taskUUID, categoryid, sessionid, clickCount)
          }
        }
 
        top10Session
    }
 
    // 写入 MySQL 数据库
    import sparkSession.implicits._
    top10SessionRDD.toDF().write
      .format("jdbc")
      .option("url", ConfigurationManager.config.getString(Constants.JDBC_URL))
      .option("user", ConfigurationManager.config.getString(Constants.JDBC_USER))
      .option("password", ConfigurationManager.config.getString(Constants.JDBC_PASSWORD))
      .option("dbtable", "top10_session")
      .mode(SaveMode.Append)
      .save()
  }
 
  // ******************** 需求三：Top10 热门品类统计 ********************
 
  // ******************** 需求二：Session 随机抽取 ********************
 
  // ******************** 需求一：Session 各范围访问步长、访问时长占比统计 ********************
}

5.4.8 需求3、四实现思路整理

5.5 需求五：页面单跳转化率统计

5.5.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 的页面单跳转化率，咱们记为 C；那么页面 5-7 的转化率怎么求呢？先须要求出符合条件的 Session 中访问页面 5 又紧接着访问了页面 7 的次数为 D，那么 D/B 即为 5-7 的单跳转化率。
  产品经理，能够根据这个指标，去尝试分析整个网站、产品各个页面的表现怎么样，是否是须要去优化产品的布局；吸引用户最终能够进入最后的支付页面。
  数据分析师，能够此数据作更深一步的计算和分析。
  企业管理层， 能够看到整个公司的网站，各个页面的之间的跳转的表现如何，能够适当调整公司的经营战略或策略。
  在如下模块中，须要根据查询对象中设置的 Session 过滤条件，先将对应的 Session 过滤出来，而后根据查询对象中设置的页面路径，计算页面单跳转化率，好比查询的页面路径为：三、五、七、8，那么就要计算 3-五、5-七、7-8 的页面单跳转化率。须要注意的一点是，页面的访问是有前后的。

5.5.2 数据源解析

动做表

5.5.3 数据结构解析

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

5.5.4 需求实现简要流程

举例

如何作

5.5.5 需求实现详细流程

略

5.5.6 MySQL 存储结构解析

-- ----------------------------
--  Table structure for `page_split_convert_rate`
-- ----------------------------
DROP TABLE IF EXISTS `page_split_convert_rate`;
CREATE TABLE `page_split_convert_rate` (
  `taskid` varchar(255) DEFAULT NULL,
  `convertRate` varchar(255) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

5.5.7 代码解析

在 analyse 中新建子模块 page，配置 pom.xml 文件，添加 scala 框架的支持

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>analyse</artifactId>
        <groupId>com.atguigu</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
 
    <artifactId>page</artifactId>
 
    <dependencies>
        <dependency>
            <groupId>com.atguigu</groupId>
            <artifactId>commons</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <!-- Spark 的依赖引入 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
        </dependency>
        <!-- 引入 Scala -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <!-- scala-maven-plugin 插件用于在任意的 maven 项目中对 scala 代码进行编译/测试/运行/文档化 -->
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.atguigu.page.PageOneStepConvertRate</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

示例代码：

package com.atguigu.page
 
import java.util.UUID
 
import commons.conf.ConfigurationManager
import commons.constant.Constants
import commons.model.UserVisitAction
import commons.utils.{DateUtils, ParamUtils}
import net.sf.json.JSONObject
import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.storage.StorageLevel
 
import scala.collection.mutable
 
object PageConvertStat {
 
  def main(args: Array[String]): Unit = {
    // 获取过滤条件，【为了方便，直接从配置文件中获取，企业中会从一个调度平台获取】
    val jsonStr = ConfigurationManager.config.getString(Constants.TASK_PARAMS)
    // 获取过滤条件对应的 JsonObject 对象
    val taskParam = JSONObject.fromObject(jsonStr)
 
    // 建立全局惟一的主键，每次执行 main 函数都会生成一个独一无二的 taskUUID，来区分不一样任务，做为写入 MySQL 数据库中那张表的主键
    val taskUUID = UUID.randomUUID().toString
 
    // 建立 sparkConf
    val sparkConf = new SparkConf().setAppName("pageConvert").setMaster("local[*]")
 
    // 建立 sparkSession（包含 sparkContext）
    val sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()
 
    // ******************** 需求五：页面单跳转化率统计 ********************
 
 
    // 获取原始的动做表数据（带有过滤条件）
    // actionRDD: RDD[UserVisitAction]
    val actionRDD = getOriActionRDD(sparkSession, taskParam)
 
    // 将用户行为信息转换为 K-V 结构
    // sessionId2ActionRDD: RDD[(sessionId, UserVisitAction)]
    val sessionId2ActionRDD = actionRDD.map(item => (item.session_id, item))
 
    // 将数据进行内存缓存
    sessionId2ActionRDD.persist(StorageLevel.MEMORY_ONLY)
 
    // 目标页面切片：将页面流路径转换为页面切片
    // targetPageFlowStr:"1,2,3,4,5,6,7"
    val targetPageFlowStr = ParamUtils.getParam(taskParam, Constants.PARAM_TARGET_PAGE_FLOW)
    // targetPageFlowArray: Array[Long][1,2,3,4,5,6,7]
    val targetPageFlowArray = targetPageFlowStr.split(",")
    // targetPageFlowArray.slice(0, targetPageFlowArray.length - 1): [1,2,3,4,5,6]
    // targetPageFlowArray.tail: [2,3,4,5,6,7]
    // targetPageFlowArray.slice(0, targetPageFlowArray.length - 1).zip(targetPageFlowArray.tail): [(1,2),(2,3),(3,4),(4,5),(5,6),(6,7)]
    val targetPageSplit = targetPageFlowArray.slice(0, targetPageFlowArray.length - 1).zip(targetPageFlowArray.tail).map {
      case (page1, page2) =>
        (page1 + "_" + page2)
    }
 
    // 获取实际页面切片
    // 对 <sessionId,访问行为> RDD，作一次 groupByKey 操做，生成页面切片
    val session2GroupActionRDD = sessionId2ActionRDD.groupByKey()
 
    // realPageSplitNumRDD: RDD[(String, 1L)]
    val realPageSplitNumRDD = session2GroupActionRDD.flatMap {
      case (sessionId, iterableUserVisitAction) =>
        // item1: UserVisitAction
        // item2: UserVisitAction
        // sortList: List[UserVisitAction] // 排好序的 UserVisitAction
        val sortList = iterableUserVisitAction.toList.sortWith((item1, item2) => {
          DateUtils.parseTime(item1.action_time).getTime < DateUtils.parseTime(item2.action_time).getTime
        })
 
        // 获取 page 信息
        // pageList: List[Long]
        val pageList = sortList.map {
          case userVisitAction =>
            userVisitAction.page_id
        }
 
        // pageList.slice(0, pageList.length - 1): List[1,2,3,...,N-1]
        // pageList.tail: List[2,3,4,...,N]
        // pageList.slice(0, pageList.length - 1).zip(pageList.tail): List[(1,2),(2,3),(3,4),...,(N-1,N)]
        val realPageSplit = pageList.slice(0, pageList.length - 1).zip(pageList.tail).map {
          case (page1, page2) =>
            (page1 + "_" + page2)
        }
 
        // 过滤：留下存在于 targetPageSplit 中的页面切片
        val realPageSplitFilter = realPageSplit.filter {
          case realPageSplit =>
            targetPageSplit.contains(realPageSplit)
        }
 
        realPageSplitFilter.map {
          case realPageSplitFilter =>
            (realPageSplitFilter, 1L)
        }
    }
 
    // 聚合
    // realPageSplitCountMap; Map[(page1_page2, count)]
    val realPageSplitCountMap = realPageSplitNumRDD.countByKey()
 
    realPageSplitCountMap.foreach(println(_))
 
    val startPage = targetPageFlowArray(0).toLong
 
    val startPageCount = sessionId2ActionRDD.filter {
      case (sessionId, userVisitAction) =>
        userVisitAction.page_id == startPage.toLong
    }.count()
 
    println("哈啊哈"+ startPageCount)
 
    // 获得最后的统计结果
    getPageConvertRate(sparkSession, taskUUID, targetPageSplit, startPageCount, realPageSplitCountMap)
  }
 
  // ******************** 需求五：页面单跳转化率统计 ********************
 
  /**
    * 计算页面切片转化率
    *
    * @param sparkSession
    * @param taskUUID
    * @param targetPageSplit
    * @param startPageCount
    * @param realPageSplitCountMap
    */
  def getPageConvertRate(sparkSession: SparkSession,
                         taskUUID: String,
                         targetPageSplit: Array[String],
                         startPageCount: Long,
                         realPageSplitCountMap: collection.Map[String, Long]): Unit = {
    val pageSplitRatioMap = new mutable.HashMap[String, Double]()
 
    var lastPageCount = startPageCount.toDouble
 
    // 1_2,2_3,3_4,...
    for (pageSplit <- targetPageSplit) {
      // 第一次循环：currentPageSplitCount: page1_page2   lastPageCount: page1
      // 第二次循环：currentPageSplitCount: page2_page3   lastPageCount: page1_page2
      val currentPageSplitCount = realPageSplitCountMap.get(pageSplit).get.toDouble
      val rate = currentPageSplitCount / lastPageCount
      pageSplitRatioMap.put(pageSplit, rate)
      lastPageCount = currentPageSplitCount
    }
 
    val convertRateStr = pageSplitRatioMap.map {
      case (pageSplit, rate) =>
        pageSplit + "=" + rate
    }.mkString("|")
 
 
    // 封装数据
    val pageSplitConvertRate = PageSplitConvertRate(taskUUID, convertRateStr)
 
    val pageSplitConvertRateRDD = sparkSession.sparkContext.makeRDD(Array(pageSplitConvertRate))
 
    // 写入到 MySQL
    import sparkSession.implicits._
    pageSplitConvertRateRDD.toDF().write
      .format("jdbc")
      .option("url", ConfigurationManager.config.getString(Constants.JDBC_URL))
      .option("dbtable", "page_split_convert_rate")
      .option("user", ConfigurationManager.config.getString(Constants.JDBC_USER))
      .option("password", ConfigurationManager.config.getString(Constants.JDBC_PASSWORD))
      .mode(SaveMode.Append)
      .save()
  }
 
  /**
    * 根据日期范围获取对象的用户行为数据
    *
    * @param sparkSession
    * @param taskParam
    * @return
    */
  def getOriActionRDD(sparkSession: SparkSession, taskParam: JSONObject) = {
    // 先获取所用到的过滤条件：开始日期 和 结束日期
    val startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE)
    val endDate = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE)
    // 把全部的时间范围在 startDate 和 endDate 之间的数据查询出来
    val sql = "select * from user_visit_action where date>='" + startDate + "' and date<='" + endDate + "'"
 
    // 在对 DataFrame 和 Dataset 进行许多操做都须要这个包进行支持
    import sparkSession.implicits._
    sparkSession.sql(sql).as[UserVisitAction].rdd // DataFrame(Row类型) -> DataSet(样例类类型) -> rdd(样例类)
  }
}

5.5.8 需求五实现思路整理

5.6 需求六：各区域 Top3 商品统计

5.6.1 需求解析

根据用户指定的日期查询条件范围，统计各个区域下的最热门【点击】的 top3 商品，区域信息、各个城市的信息在项目中用固定值进行配置，由于不怎么变更。

1、查询 task，获取日期范围，经过 Spark SQL，查询 user_visit_action 表中的指定日期范围内的数据，过滤出商品点击行为，click_product_id is not null、click_product_id != 'NULL'、click_product_id != 'null'、city_id、click_product_id。
2、使用 Spark SQL 从 MySQL 中查询出来城市信息（city_id、city_name、area），用户访问行为数据要跟城市信息进行 join， city_id、city_name、area、product_id， 将 RDD 转换成 DataFrame，注册成一个临时表。
3、Spark SQL 内置函数（case when），对 area 打标记（华东大区，A 级；华中大区，B 级；东北大区，C 级；西北大区，D 级），area_level。
4、计算出来每一个区域下每一个商品的点击次数，group by area, product_id 保留每一个区域的城市名称列表；自定义 UDAF，group_concat_distinct()函数，聚合出来一个 city_names 字段，area、product_id、city_names、click_count。
5、join 商品明细表，hive（product_id、product_name、extend_info），extend_info 是 json 类型；自定义 UDF，get_json_object() 函数，取出其中的 product_status 字段， if() 函数（Spark SQL 内置函数），判断： 0 自营，1 第三方（area、product_id、 city_names、click_count、product_name、product_status）。
6、开窗函数，根据 area 来聚合，获取每一个 area 下，click_count 排名前 3 的 product 信息：area、area_level、product_id、city_names、click_count、product_name、product_status。
7、结果写入 MySQL 表中。

5.6.2 数据源解析

商品信息表 + 用户访问行为表

5.6.3 数据结构解析

城市信息（城市 ID，城市名称，区域名称）

Array((0L, "北京", "华北"), (1L, "上海", "华东"), (2L, "南京", "华东"), (3L, "广州", "华南"), (4L, "三亚", "华南"), (5L, "武汉", "华中"), (6L, "长沙", "华中"), (7L, "西安", "西北"), (8L, "成都", "西南"), (9L, "哈尔滨", "东北"))

5.6.4 需求实现简要流程

5.6.5 需求实现详细流程

略

5.6.6 MySQL 存储结构解析

-- ----------------------------
--  Table structure for `area_top3_product`
-- ----------------------------
DROP TABLE IF EXISTS `area_top3_product`; 
CREATE TABLE `area_top3_product` (
  `taskid` varchar(255) DEFAULT NULL,
  `area` varchar(255) DEFAULT NULL,
  `areaLevel` varchar(255) DEFAULT NULL,
  `productid` int(11) DEFAULT NULL,
  `cityInfos` varchar(255) DEFAULT NULL,
  `clickCount` int(11) DEFAULT NULL,
  `productName` varchar(255) DEFAULT NULL,
  `productStatus` varchar(255) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

5.6.7 代码解析

在 analyse 中新建子模块 product，配置 pom.xml 文件，添加 scala 框架的支持

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>analyse</artifactId>
        <groupId>com.atguigu</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
 
    <artifactId>product</artifactId>
 
    <dependencies>
        <dependency>
            <groupId>com.atguigu</groupId>
            <artifactId>commons</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <!-- Spark 的依赖引入 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
        </dependency>
        <!-- 引入 Scala -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
        </dependency>
    </dependencies>
 
    <build>
        <plugins>
            <plugin>
                <!-- scala-maven-plugin 插件用于在任意的 maven 项目中对 scala 代码进行编译/测试/运行/文档化 -->
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.atguigu.product.AreaTop3ProductApp</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

示例代码：

package com.atguigu.product
 
import java.util.UUID
 
import commons.conf.ConfigurationManager
import commons.constant.Constants
import commons.utils.ParamUtils
import net.sf.json.JSONObject
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SaveMode, SparkSession}
 
object AreaTop3ProductStat {
 
  def main(args: Array[String]): Unit = {
    // 获取过滤条件，【为了方便，直接从配置文件中获取，企业中会从一个调度平台获取】
    val jsonStr = ConfigurationManager.config.getString(Constants.TASK_PARAMS)
    // 获取过滤条件对应的 JsonObject 对象
    val taskParam = JSONObject.fromObject(jsonStr)
 
    // 建立全局惟一的主键，每次执行 main 函数都会生成一个独一无二的 taskUUID，来区分不一样任务，做为写入 MySQL 数据库中那张表的主键
    val taskUUID = UUID.randomUUID().toString
 
    // 建立 sparkConf
    val sparkConf = new SparkConf().setAppName("product").setMaster("local[*]")
 
    // 建立 sparkSession（包含 sparkContext）
    val sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()
 
    // ******************** 需求六：各区域 Top3 商品统计 ********************
 
    // 获取 用户访问行为的数据 (cityId, clickProductId)
    // cityId2ProductIdRDD: RDD[(cityId, clickProductId)]
    val cityId2ProductIdRDD = getCityAndProductInfo(sparkSession, taskParam)
 
    // 获取 城市信息 (城市 ID，城市名称，区域名称)
    // cityId2AreaInfoRDD: RDD[(cityId, CityAreaInfo)]
    val cityId2AreaInfoRDD = getCityAreaInfo(sparkSession)
 
    // 作 Join 操做，获得 (city_id, city_name, area, click_product_id)
    // 临时表 temp_area_product_info 中的一条数据就表明一次点击商品的行为
    getAreaProductIdBasicInfoTable(sparkSession, cityId2ProductIdRDD, cityId2AreaInfoRDD)
 
    // 自定义 UDF 函数：实现字符串带去重的拼接
    sparkSession.udf.register("concat_long_string", (v1: Long, v2: String, split: String) => {
      v1 + split + v2
    })
    sparkSession.udf.register("group_concat_distinct", new GroupConcatDistinct)
 
    // 统计 每个区域里每个商品被点击的次数，获得 (area, click_product_id, click_count, city_infos)
    getAreaProductClickCountTable(sparkSession)
 
    // 自定义 UDAF 函数：实现从 json 串中取出指定字段的值
    sparkSession.udf.register("get_json_field", (json: String, field: String) => {
      val jsonObject = JSONObject.fromObject(json)
      jsonObject.getString(field)
    })
 
    // 将 temp_area_product_count 表 join 商品信息表 product_info
    getAreaProductClickCountInfo(sparkSession)
 
    // 获取 各区域 Top3 商品（使用到了开窗函数）
    getAreaTop3Product(sparkSession, taskUUID)
 
    // 测试
    // sparkSession.sql("select * from temp_area_product_info").show
    // sparkSession.sql("select * from temp_area_product_count").show
    // sparkSession.sql("select * from temp_area_count_product_info").show
    // sparkSession.sql("select * from temp_test").show
  }
 
  /**
    * 获取 各区域 Top3 商品（使用了开窗函数）
    *
    * @param sparkSession
    * @param taskUUID
    * @return
    */
  def getAreaTop3Product(sparkSession: SparkSession, taskUUID: String) = {
//    val sql = "select area, city_infos, click_product_id, click_count, product_name, product_status, " +
//      "row_number() over(partition by area order by click_count desc) row_number from temp_area_count_product_info"
//    sparkSession.sql(sql).createOrReplaceTempView("temp_test") // 测试
 
    val sql = "select area, " +
      "case " +
      "when area='华北' or area='华东' then 'A_Level' " +
      "when area='华中' or area='华男' then 'B_Level' " +
      "when area='西南' or area='西北' then 'C_Level' " +
      "else 'D_Level' " +
      "end area_level, " +
      "city_infos, click_product_id, click_count, product_name, product_status from (" +
      "select area, city_infos, click_product_id, click_count, product_name, product_status, " +
      "row_number() over(partition by area order by click_count desc) row_number from temp_area_count_product_info) t where row_number <= 3"
    val areaTop3ProductRDD = sparkSession.sql(sql).rdd.map {
      case row =>
        AreaTop3Product(taskUUID, row.getAs[String]("area"), row.getAs[String]("area_level"),
          row.getAs[Long]("click_product_id"), row.getAs[String]("city_infos"),
          row.getAs[Long]("click_count"), row.getAs[String]("product_name"),
          row.getAs[String]("product_status"))
    }
 
    import sparkSession.implicits._
    areaTop3ProductRDD.toDF().write
      .format("jdbc")
      .option("url", ConfigurationManager.config.getString(Constants.JDBC_URL))
      .option("dbtable", "area_top3_product")
      .option("user", ConfigurationManager.config.getString(Constants.JDBC_USER))
      .option("password", ConfigurationManager.config.getString(Constants.JDBC_PASSWORD))
      .mode(SaveMode.Append)
      .save()
  }
 
  /**
    * 将 temp_area_product_count 表 join 商品信息表
    *
    * @param sparkSession
    */
  def getAreaProductClickCountInfo(sparkSession: SparkSession): Unit = {
    // temp_area_product_count: (area, click_product_id, click_count, city_infos) tapc
    // product_info: (product_id, product_name, extend_info)  pi
    val sql = "select tapc.area, tapc.city_infos, tapc.click_product_id, tapc.click_count, pi.product_name, " +
      "if (get_json_field(pi.extend_info, 'product_status') = '0', 'Self', 'Third Party') product_status" +
      " from temp_area_product_count tapc join product_info pi on tapc.click_product_id = pi.product_id"
 
    sparkSession.sql(sql).createOrReplaceTempView("temp_area_count_product_info")
  }
 
  /**
    * 统计 每个区域里每个商品被点击的次数
    */
  def getAreaProductClickCountTable(sparkSession: SparkSession): Unit = {
    val sql = "select area, click_product_id, count(*) click_count," +
      " group_concat_distinct(concat_long_string(city_id, city_name, ':')) city_infos" +
      " from temp_area_product_info group by area, click_product_id"
 
    sparkSession.sql(sql).createOrReplaceTempView("temp_area_product_count")
  }
 
  /**
    * 将 用户访问行为的数据 (cityId, clickProductId) 和 城市信息 (城市 ID，城市名称，区域名称) 作 join 操做，获得所需的临时表数据
    *
    * @param sparkSession
    * @param cityId2ProductIdRDD
    * @param cityId2AreaInfoRDD
    */
  def getAreaProductIdBasicInfoTable(sparkSession: SparkSession,
                                     cityId2ProductIdRDD: RDD[(Long, Long)],
                                     cityId2AreaInfoRDD: RDD[(Long, CityAreaInfo)]): Unit = {
    val areaProductIdBasicInfoRDD = cityId2ProductIdRDD.join(cityId2AreaInfoRDD).map {
      case (cityId, (clickProductId, cityAreaInfo)) =>
        (cityId, cityAreaInfo.city_name, cityAreaInfo.area, clickProductId)
    }
 
    import sparkSession.implicits._
    // 转换为临时表的时候须要指定字段的名称
    areaProductIdBasicInfoRDD.toDF("city_id", "city_name", "area", "click_product_id").createOrReplaceTempView("temp_area_product_info")
  }
 
  /**
    * 获取 城市信息（城市 ID，城市名称，区域名称）
    *
    * @param sparkSession
    */
  def getCityAreaInfo(sparkSession: SparkSession) = {
    val cityAreaInfoArray = Array((0L, "北京", "华北"), (1L, "上海", "华东"), (2L, "南京", "华东"),
      (3L, "广州", "华南"), (4L, "三亚", "华南"), (5L, "武汉", "华中"),
      (6L, "长沙", "华中"), (7L, "西安", "西北"), (8L, "成都", "西南"),
      (9L, "哈尔滨", "东北"))
 
    // RDD[(cityId, CityAreaInfo)]
    sparkSession.sparkContext.makeRDD(cityAreaInfoArray).map {
      case (cityId, cityName, area) =>
        (cityId, CityAreaInfo(cityId, cityName, area))
    }
  }
 
  /**
    * 获取 用户访问行为的数据 (city_id, click_product_id)
    *
    * @param sparkSession
    * @param taskParam
    * @return
    */
  def getCityAndProductInfo(sparkSession: SparkSession, taskParam: JSONObject) = {
    val startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE)
    val endDate = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE)
 
    // 只获取发生过点击的 action 的数据，获取到的一条 action 数据就表明一个点击行为
    val sql = "select city_id, click_product_id from user_visit_action where date>='" + startDate +
      "' and date<='" + endDate + "' and click_product_id != -1"
 
    import sparkSession.implicits._
    sparkSession.sql(sql).as[CityClickProduct].rdd.map {
      case cityIdAndProductId =>
        (cityIdAndProductId.city_id, cityIdAndProductId.click_product_id)
    }
  }
}

自定义弱类型的 UDAF 函数

package com.atguigu.product
 
import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, StringType, StructField, StructType}
 
/**
  * 自定义弱类型的 UDAF 函数
  */
class GroupConcatDistinct extends UserDefinedAggregateFunction {
  // 设置 UDAF 函数的输入类型为 String
  override def inputSchema: StructType = StructType(StructField("cityInfoInput", StringType) :: Nil)
 
  // 设置 UDAF 函数的缓冲区类型为 String
  override def bufferSchema: StructType = StructType(StructField("cityInfoBuffer", StringType) :: Nil)
 
  // 设置 UDAF 函数的输出类型为 String
  override def dataType: DataType = StringType
 
  // 设置 UDAF 函数的输入数据和输出数据是一致的
  override def deterministic: Boolean = true
 
  // 初始化自定义的 UDAF 函数
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = ""
  }
 
  // 设置 UDAF 函数的缓冲区更新：实现一个字符串带去重的拼接
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    var cityInfoBuffer = buffer.getString(0)
    val cityInfoInput = input.getString(0)
 
    if (!cityInfoBuffer.contains(cityInfoInput)) {
      if ("".equals(cityInfoBuffer)) {
        cityInfoBuffer += cityInfoInput
      } else {
        cityInfoBuffer += "," + cityInfoInput
      }
    }
 
    buffer.update(0, cityInfoBuffer)
  }
 
  // 把两个自定义的 UDAF 函数的值合并在一块儿
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    // cityInfoBuffer1: cityId1:cityName1, cityId2:cityName2, cityId3:cityName3, ...
    var cityInfoBuffer1 = buffer1.getString(0)
    // cityInfoBuffer2: cityId1:cityName1, cityId2:cityName2, cityId3:cityName3, ...
    val cityInfoBuffer2 = buffer2.getString(0)
 
    // 将 cityInfoBuffer2 中的数据带去重的加入到 cityInfoBuffer1 中
    for (cityInfo <- cityInfoBuffer2.split(",")) {
      if (!cityInfoBuffer1.contains(cityInfo)) {
        if ("".equals(cityInfoBuffer1)) {
          cityInfoBuffer1 += cityInfo
        } else {
          cityInfoBuffer1 += "," + cityInfo
        }
      }
    }
 
    buffer1.update(0, cityInfoBuffer1)
  }
 
  // 返回结果
  override def evaluate(buffer: Row): Any = {
    buffer.getString(0)
  }
}

AreaTop3Product 类

/**
  * 各省 top3 热门广告
  *
  * @author
  *
  */
case class AdProvinceTop3(date: String,
                          province: String,
                          adid: Long,
                          clickCount: Long)

5.6.8 需求六实现思路整理

5.7 需求七：广告点击黑名单实时统计

广告流量实时统计【概览】

5.7.1 需求解析

  实现实时的动态黑名单机制：将天天对某个广告点击超过 100 次的用户拉黑。

5.7.2 数据源解析

Kafka 数据：timestamp province city userid adid

5.7.3 数据结构解析

略

5.7.4 需求实现简要流程

5.7.5 需求实现详细流程

略

5.7.6 MySQL 存储结构解析

-- ----------------------------
--  Table structure for `ad_blacklist`
-- ----------------------------
DROP TABLE IF EXISTS `ad_blacklist`; 
CREATE TABLE `ad_blacklist` (
  `userid` int(11) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
 
-- ----------------------------
--  Table structure for `ad_user_click_count`
-- ----------------------------
DROP TABLE IF EXISTS `ad_user_click_count`; 
CREATE TABLE `ad_user_click_count` (
  `date` varchar(30) DEFAULT NULL,
  `userid` int(11) DEFAULT NULL,
  `adid` int(11) DEFAULT NULL,
  `clickCount` int(11) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

5.7.7 代码解析

在 analyse 中新建子模块 advertising，配置 pom.xml 文件，添加 scala 框架的支持

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>analyse</artifactId>
        <groupId>com.atguigu</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
 
    <artifactId>advertising</artifactId>
 
    <dependencies>
        <dependency>
            <groupId>com.atguigu</groupId>
            <artifactId>commons</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
 
        <!-- Spark 的依赖引入 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
        </dependency>
 
        <!-- 引入 Scala -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
        </dependency>
 
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
    </dependencies>
 
    <build>
        <plugins>
            <plugin>
                <!-- scala-maven-plugin 插件用于在任意的 maven 项目中对 scala 代码进行编译/测试/运行/文档化 -->
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.atguigu.stream.AdClickRealTimeStat</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

准备工做：数据生成与数据消费测试

一、先启动 zookeeper 集群，再启动 kafka 集群

[atguigu@hadoop102 ~]$ zkstart.sh
[atguigu@hadoop102 ~]$ kafka-start.sh 

二、启动一个 kafka 消费者进程

老版本：消费者会将本身的 offset 文件保存在 zookeeper(低版本的kafka)。因此消费者链接的是 zookeeper。
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic AdRealTimeLog1
 
新版本：消费者会将本身的 offset 文件保存在 kafka 集群中(高版本的kafka)。因此消费者链接的是 kafka。这样作的好处是：提升了效率，减小了网络传输。
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic AdRealTimeLog1

三、启动数据生成程序 MockRealTimeData

package com.atguigu.stream
 
import commons.conf.ConfigurationManager
import commons.constant.Constants
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
 
object AdClickRealTimeStat {
 
  def main(args: Array[String]): Unit = {
    // 构建 Spark 上下文
    val sparkConf = new SparkConf().setAppName("streamingRecommendingSystem").setMaster("local[*]")
 
    // 建立 Spark 客户端
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    val sc = spark.sparkContext
    val ssc = new StreamingContext(sc, Seconds(5))
 
    // 设置检查点目录
    ssc.checkpoint("./streaming_checkpoint")
 
    // 获取 kafka 的配置
    val kafka_brokers = ConfigurationManager.config.getString(Constants.KAFKA_BROKERS)
    val kafka_topics = ConfigurationManager.config.getString(Constants.KAFKA_TOPICS)
 
    // kafka 消费者参数配置
    val kafkaParam = Map(
      "bootstrap.servers" -> kafka_brokers, // 用于初始化链接到集群的地址
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "commerce-consumer-group", // 用于标识这个消费者属于哪一个消费团体(组)
      "auto.offset.reset" -> "latest", // 若是没有初始化偏移量或者当前的偏移量不存在任何服务器上，可使用这个配置属性，latest 表示自动重置偏移量为最新的偏移量
      "enable.auto.commit" -> (false: java.lang.Boolean) // 若是是 true，则这个消费者的偏移量会在后台自动提交
    )
 
    // 建立 DStream，返回接收到的输入数据
    // LocationStrategies：                  根据给定的主题和集群地址建立 consumer
    // LocationStrategies.PreferConsistent： 持续的在全部 Executor 之间均匀分配分区(即把 Executor 当成 Kafka Consumer)，经常使用该方式
    // ConsumerStrategies：                  选择如何在 Driver 和 Executor 上建立和配置 Kafka Consumer
    // ConsumerStrategies.Subscribe：        订阅一系列主题
    // adRealTimeLogDStream: DStream[RDD, RDD, RDD, ...] -> RDD[Message] -> Message: key value
    val adRealTimeLogDStream = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Array(kafka_topics), kafkaParam))
 
    // 取出 DStream 里面每一条数据的 value 值
    // adRealTimeLogDStream: DStream[RDD, RDD, RDD, ...] -> RDD[String] -> String: timestamp province city userid adid
    val adRealTimeValueDStream = adRealTimeLogDStream.map(consumerRecordRDD => consumerRecordRDD.value())
 
    // 用于 Kafka Stream 的线程非安全问题，从新分区切断血统
    // adRealTimeValueDStream = adRealTimeValueDStream.repartition(400)
 
    // 根据黑名单进行数据过滤
    // 刚刚接受到原始的用户点击行为日志以后
    // 根据 mysql 中的动态黑名单，进行实时的黑名单过滤（黑名单用户的点击行为，直接过滤掉，不要了）
    // 使用 transform 算子（将 dstream 中的每一个 batch RDD 进行处理，转换为任意的其余 RDD，功能很强大）
    val adRealTimeFilterDStream = adRealTimeValueDStream.transform {
      consumerRecordRDD =>
        // 首先从 MySQL 中查询全部黑名单用户
        // adBlacklists: Array[AdBlacklist]  AdBlacklist: userId
        val adBlacklistArray = AdBlacklistDAO.findAll()
 
        // userIdArray: Array[Long]  [userId1, userId2, ...]
        val userIdArray = adBlacklistArray.map(item => item.userid)
 
        consumerRecordRDD.filter {
          // consumerRecord: timestamp province city userid adid
          case consumerRecord =>
            val consumerRecordSplit = consumerRecord.split(" ")
            val userId = consumerRecordSplit(3).toLong
            !userIdArray.contains(userId)
        }
    }
 
    adRealTimeFilterDStream.foreachRDD(rdd => rdd.foreach(println(_)))
 
    ssc.start()
    ssc.awaitTermination()
  }
}

四、启动数据消费程序 AdClickRealTimeStat

import java.util.Properties
 
import commons.conf.ConfigurationManager
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
 
import scala.collection.mutable.ArrayBuffer
import scala.util.Random
 
object MockRealTimeData {
 
  def main(args: Array[String]): Unit = {
    // 获取配置文件 commerce.properties 中的 Kafka 配置参数
    val broker = ConfigurationManager.config.getString("kafka.broker.list")
    val topic = ConfigurationManager.config.getString("kafka.topics")
 
    // 建立 Kafka 生产者
    val kafkaProducer = createKafkaProducer(broker)
 
    while (true) {
      // 随机产生实时数据并经过 Kafka 生产者发送到 Kafka 集群中
      for (item <- generateMockData()) {
        kafkaProducer.send(new ProducerRecord[String, String](topic, item))
      }
      Thread.sleep(5000)
    }
  }
 
  /**
    * 实时模拟数据的生成
    *
    * 时间点: 当前时间毫秒
    * userId: 0 - 99
    * 省份、城市 ID 相同: 1 - 9
    * adid: 0 - 19
    * ((0L,"北京","北京"),(1L,"上海","上海"),(2L,"南京","江苏省"),(3L,"广州","广东省"),(4L,"三亚","海南省"),(5L,"武汉","湖北省"),(6L,"长沙","湖南省"),(7L,"西安","陕西省"),(8L,"成都","四川省"),(9L,"哈尔滨","东北省"))
    *
    * 格式 ：timestamp province city userid adid
    *       某个时间点 某个省份 某个城市 某个用户 某个广告
    */
  def generateMockData(): Array[String] = {
    val array = ArrayBuffer[String]()
    val random = new Random()
 
    // 模拟实时数据：timestamp province city userid adid
    for (i <- 0 until 50) {
      val timestamp = System.currentTimeMillis()
      val province = random.nextInt(10)
      val city = province
      val adid = random.nextInt(20)
      val userid = random.nextInt(100)
 
      // 拼接实时数据
      array += timestamp + " " + province + " " + city + " " + userid + " " + adid
    }
 
    array.toArray
  }
 
  def createKafkaProducer(broker: String): KafkaProducer[String, String] = {
    // 建立配置对象
    val prop = new Properties()
    // 添加配置
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker)
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
 
    // 根据配置建立 Kafka 生产者
    new KafkaProducer[String, String](prop)
  }
}

五、查看控制台消费者的输出 以及 Linux 上的 kafka 消费者的输出，没有问题！下面实现具体的需求。

业务代码

  /**
    * 需求七：广告点击黑名单实时统计
    *
    * @param adRealTimeFilterDStream
    */
  def generateBlackListStat(adRealTimeFilterDStream: DStream[String]) = {
    // adRealTimeFilterDStream: DStream[RDD, RDD, RDD, ...] -> RDD[String] -> String: timestamp province city userid adid
    // key2NumDStream: DStream[RDD, RDD, RDD, ...] -> RDD[(key, 1L)]
    val key2NumDStream = adRealTimeFilterDStream.map {
      // consumerRecordRDD: timestamp province city userid adid
      case consumerRecordRDD =>
        val consumerRecordSplit = consumerRecordRDD.split(" ")
        val timestamp = consumerRecordSplit(0).toLong
        // dateKey: yyyyMMdd
        val dateKey = DateUtils.formatDateKey(new Date(timestamp))
        val userid = consumerRecordSplit(3).toLong
        val adid = consumerRecordSplit(4).toLong
 
        val key = dateKey + "_" + userid + "_" + adid // 组合 key
 
        (key, 1L)
    }
 
    // key2CountDStream: DStream[RDD, RDD, RDD, ...] -> RDD[(key, 25)]
    val key2CountDStream = key2NumDStream.reduceByKey(_ + _)
 
    // 根据每个 RDD 里面的数据，更新 用户点击次数表 数据
    key2CountDStream.foreachRDD {
      rdd => rdd.foreachPartition {
        items =>
          val clickCountArray = new ArrayBuffer[AdUserClickCount]()
 
          for ((key, count) <- items) {
            // 切割数据
            val keySplit = key.split("_")
 
            // 取出数据
            val date = keySplit(0)
            val userid = keySplit(1).toLong
            val adid = keySplit(2).toLong
 
            // 封装数据，并放入 ArrayBuffer 中
            clickCountArray += AdUserClickCount(date, userid, adid, count)
          }
 
          // 更新 MySQl 数据库中表的数据
          AdUserClickCountDAO.updateBatch(clickCountArray.toArray)
      }
    }
 
    // 对 DStream 作 filter 操做：就是遍历 DStream 中的每个 RDD 中的每一条数据
    // key2BlackListDStream: DStream[RDD, RDD, RDD, ...] -> RDD[(key, 150)]
    val key2BlackListDStream = key2CountDStream.filter {
      case (key, count) =>
        // 切割数据
        val keySplit = key.split("_")
 
        // 取出数据
        val date = keySplit(0)
        val userid = keySplit(1).toLong
        val adid = keySplit(2).toLong
 
        val clickCount = AdUserClickCountDAO.findClickCountByMultiKey(date, userid, adid)
 
        if (clickCount > 100) {
          true // 留下
        } else {
          false // 过滤掉
        }
    }
 
    // key2BlackListDStream.map: DStream[RDD[userid]]
    val userIdDStream = key2BlackListDStream.map {
      case (key, count) =>
        key.split("_")(1).toLong
    }.transform(rdd => rdd.distinct()) // 转换 key 并去重
 
    // 将结果批量插入 MySQL 数据库中
    userIdDStream.foreachRDD {
      rdd => rdd.foreachPartition {
        items =>
          val userIdArray = new ArrayBuffer[AdBlacklist]()
 
          for (userId <- items) {
            userIdArray += AdBlacklist(userId)
          }
 
          AdBlacklistDAO.insertBatch(userIdArray.toArray)
      }
    }
  }

AdBlacklist 类

/**
  * 广告黑名单
  *
  * @author
  *
  */
case class AdBlacklist(userid: Long)
 
/**
  * 用户广告点击量
  *
  * @author
  *
  */
case class AdUserClickCount(date: String,
                            userid: Long,
                            adid: Long,
                            clickCount: Long)

5.7.8 需求七实现思路整理

略

5.8 需求八：各省各城市广告点击量实时统计

5.8.1 需求解析

略

5.8.2 数据源解析

Kafka 数据：timestamp province city userid adid

5.8.3 数据结构解析

略

5.8.4 需求实现简要流程

5.8.5 需求实现详细流程

略

5.8.6 MySQL 存储结构解析

-- ----------------------------
--  Table structure for `ad_stat`
-- ----------------------------
DROP TABLE IF EXISTS `ad_stat`; 
CREATE TABLE `ad_stat` (
  `date` varchar(30) DEFAULT NULL,
  `province` varchar(100) DEFAULT NULL,
  `city` varchar(100) DEFAULT NULL,
  `adid` int(11) DEFAULT NULL,
  `clickCount` int(11) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

5.8.7 代码解析

业务代码

  /**
    * 需求八：各省各城市广告点击量实时统计（使用累计统计）
    *
    * @param adRealTimeFilterDStream
    */
  def provinceCityClickStat(adRealTimeFilterDStream: DStream[String]) = {
    // adRealTimeFilterDStream: DStream[RDD, RDD, RDD, ...] -> RDD[String] -> String: timestamp province city userid adid
    // key2NumDStream: DStream[RDD, RDD, RDD, ...] -> RDD[(key, 1L)]
    val key2NumDStream = adRealTimeFilterDStream.map {
      case consumerRecordRDD =>
        val consumerRecordSplit = consumerRecordRDD.split(" ")
        val timestamp = consumerRecordSplit(0).toLong
        // dateKey: yyyyMMdd
        val dateKey = DateUtils.formatDateKey(new Date(timestamp))
        val province = consumerRecordSplit(1)
        val city = consumerRecordSplit(2)
        val adid = consumerRecordSplit(4)
 
        val key = dateKey + "_" + province + "_" + city + "_" + adid // 组合 key
 
        (key, 1L)
    }
 
    // 执行 updateStateByKey 操做（全局的累计性的操做）
    val key2StateDStream = key2NumDStream.updateStateByKey[Long] {
      (values: Seq[Long], state: Option[Long]) =>
        var newValue = 0L
 
        if (state.isDefined) {
          newValue = state.get
        }
 
        for (value <- values) {
          newValue += value
        }
 
        Some(newValue)
    }
 
    // 将结果批量插入 MySQL 数据库中
    key2StateDStream.foreachRDD {
      rdd =>
        rdd.foreachPartition {
          items =>
            val adStatArray = new ArrayBuffer[AdStat]()
 
            for ((key, count) <- items) {
              // 切割数据
              val keySplit = key.split("_")
 
              // 取出数据
              val date = keySplit(0)
              val province = keySplit(1)
              val city = keySplit(2)
              val adid = keySplit(3).toLong
 
              // 封装数据，并放入 ArrayBuffer 中
              adStatArray += AdStat(date, province, city, adid, count)
            }
 
            AdStatDAO.updateBatch(adStatArray.toArray)
        }
    }
  }

AdStat 类

/**
  * 广告实时统计
  *
  * @author
  *
  */
case class AdStat(date: String,
                  province: String,
                  city: String,
                  adid: Long,
                  clickCount: Long)

5.8.8 需求八实现思路整理

略

5.9 需求九：天天每一个省份 Top3 热门广告

5.9.1 需求解析

略

5.9.2 数据源解析

数据来源于需求八 updateStateByKey 获得的 DStream。

5.9.3 数据结构解析

Dstream[(dateKey_province_city_adid, count)]

5.9.4 需求实现简要流程

5.9.5 需求实现详细流程

略

5.9.6 MySQL 存储结构解析

-- ----------------------------
--  Table structure for `ad_province_top3`
-- ----------------------------
DROP TABLE IF EXISTS `ad_province_top3`; 
CREATE TABLE `ad_province_top3` (
  `date` varchar(30) DEFAULT NULL,
  `province` varchar(100) DEFAULT NULL,
  `adid` int(11) DEFAULT NULL,
  `clickCount` int(11) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

5.9.7 代码解析

业务代码

  /**
    * 需求九：天天每一个省份 Top3 热门广告
    *
    * @param spark
    * @param key2ProvinceCityCountDStream
    */
  def provinceTop3AdverStat(spark: SparkSession, key2ProvinceCityCountDStream: DStream[(String, Long)]) = {
    // key2ProvinceCityCountDStream: DStream[RDD, RDD, RDD, ...] -> RDD[(key, count)]
    // key: date_province_city_adid
 
    // 转换 key
    // key2ProvinceCountDStream: DStream[RDD, RDD, RDD, ...] -> RDD[(newKey, count)]
    // newKey: date_province_adid
    val key2ProvinceCountDStream = key2ProvinceCityCountDStream.map {
      case (key, count) =>
        val keySplit = key.split("_")
        val date = keySplit(0)
        val province = keySplit(1)
        val adid = keySplit(3)
 
        val newKey = date + "_" + province + "_" + adid // 组合 key
 
        (newKey, count)
    }
 
    // 聚合新 key
    val key2ProvinceAggrCountDStream = key2ProvinceCountDStream.reduceByKey(_ + _)
 
    val top3DStream = key2ProvinceAggrCountDStream.transform {
      rdd =>
        // rdd: RDD[(newKey, count)]  newKey: date_province_adid
 
        // 转化为新的 RDD
        val basicDataRDD = rdd.map {
          case (key, count) =>
            val keySplit = key.split("_")
            val date = keySplit(0)
            val province = keySplit(1)
            val adid = keySplit(2).toLong
 
            (date, province, adid, count)
        }
 
        // 将 RDD 转化为 DF
        import spark.implicits._
        basicDataRDD.toDF("date", "province", "adid", "count").createOrReplaceTempView("temp_basic_info")
 
        // 使用 Spark SQL 执行 SQL 语句，配合开窗函数，统计出各省份 top3 热门的广告
        val sql = "select date, province, adid, count from (" +
          "select date, province, adid, count, " +
          "row_number() over(partition by date, province order by count desc) row_number from temp_basic_info) t " +
          "where row_number <= 3"
 
        spark.sql(sql).rdd
    }
 
    top3DStream.foreachRDD {
      // rdd: RDD[Row]
      rdd =>
        rdd.foreachPartition {
          // items: Row
          items =>
            val top3Array = new ArrayBuffer[AdProvinceTop3]()
 
            for (item <- items) {
              // 取出数据
              val date = item.getAs[String]("date")
              val province = item.getAs[String]("province")
              val adid = item.getAs[Long]("adid")
              val count = item.getAs[Long]("count")
 
              // 封装数据
              top3Array += AdProvinceTop3(date, province, adid, count)
            }
 
            // 写入 MySQL 数据库中
            AdProvinceTop3DAO.updateBatch(top3Array.toArray)
        }
    }
  }

AdProvinceTop3 类

/**
  * 各省 top3 热门广告
  *
  * @author
  *
  */
case class AdProvinceTop3(date: String,
                          province: String,
                          adid: Long,
                          clickCount: Long)

5.9.8 需求九实现思路整理

略

5.10 需求十：最近一小时广告点击量实时统计

5.10.1 需求解析

  统计各广告最近 1 小时内的点击量趋势：各广告最近 1 小时内各分钟的点击量。

5.10.2 数据源解析

略

5.10.3 数据结构解析

略

5.10.4 需求实现简要流程

5.10.5 需求实现详细流程

略

5.10.6 MySQL 存储结构解析

-- ----------------------------
--  Table structure for `ad_click_trend`
-- ----------------------------
DROP TABLE IF EXISTS `ad_click_trend`; 
CREATE TABLE `ad_click_trend` (
  `date` varchar(30) DEFAULT NULL,
  `hour` varchar(30) DEFAULT NULL,
  `minute` varchar(30) DEFAULT NULL,
  `adid` int(11) DEFAULT NULL,
  `clickCount` int(11) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

5.10.7 代码解析

业务代码

  /**
    * 需求十：计算最近 1 小时滑动窗口内的广告点击趋势
    *
    * @param adRealTimeFilterDStream
    * @return
    */
  def recentHourAdverClickStat(adRealTimeFilterDStream: DStream[String]) = {
    val key2TimeMinuteDStream = adRealTimeFilterDStream.map {
      // consumerRecordRDD: timestamp province city userid adid
      case consumerRecordRDD =>
        val consumerRecordSplit = consumerRecordRDD.split(" ")
        val timestamp = consumerRecordSplit(0).toLong
 
        // timeMinute: yyyyMMddHHmm
        val timeMinute = DateUtils.formatTimeMinute(new Date(timestamp))
        val adid = consumerRecordSplit(4).toLong
 
        val key = timeMinute + "_" + adid // 组合 key
 
        (key, 1L)
    }
 
    // 第一个 Minutes 表示 窗口大小，第二个 Minutes 表示 窗口滑动步长
    val key2WindowDStream = key2TimeMinuteDStream.reduceByKeyAndWindow((a: Long, b: Long) => (a + b), Minutes(60), Minutes(1))
 
    key2WindowDStream.foreachRDD {
      rdd => rdd.foreachPartition {
        // items: (key, count)
        items =>
          val trendArray = ArrayBuffer[AdClickTrend]()
 
          for ((key, count) <- items) {
            val keySplit = key.split("_")
 
            // 切割数据
            // timeMinute: yyyyMMddHHmm
            val timeMinute = keySplit(0)
 
            // 获取数据
            val date = timeMinute.substring(0, 8)   // 包头不包尾，注意是索引
            val hour = timeMinute.substring(8, 10)  // 包头不包尾，注意是索引
            val minute = timeMinute.substring(10)   // 包头不包尾，注意是索引
 
            val adid = keySplit(1).toLong
 
            // 封装数据
            trendArray += AdClickTrend(date, hour, minute, adid, count)
          }
 
          // 写入 MySQL 数据库中
          AdClickTrendDAO.updateBatch(trendArray.toArray)
      }
    }
  }

AdClickTrend 类

/**
  * 广告点击趋势
  *
  * @author
  *
  */
case class AdClickTrend(date: String,
                        hour: String,
                        minute: String,
                        adid: Long,
                        clickCount: Long)

5.10.8 需求7、8、9、十实现思路整理

第6章 项目总结

  本项目经过 Spark 技术生态栈中的 Spark Core、Spark SQL 和 Spark Streaming三个技术框架，实现了对电商平台业务的离线和实时数据统计与分析，完成了包括用户访问 session 分析、页面单跳转化率统计、热门商品离线统计、广告流量实时统计 4 个业务模块的开发工做。
  本项目涵盖了 Spark Core、Spark SQL 和 Spark Streaming 三个技术框架中核心的知识点与技术点，对于同窗们真正的理解和掌握 Spark 技术生态栈有着良好的促进做用。