0x00 教程内容

1. 项目分析
2. 编程实现

基础知识及环境准备：
1、Maven的构建，学会Scala编程
2、IDEA已经安装好了Scala插件

0x01 项目分析

1. 项目背景

我们在浏览网站的时候，我们的很多信息都会被网站的后台采集到，采集的方式有很多种，至于怎么采集，以后有机会再写教程详细说，这篇教程主要是讲解采集到了数据后，对数据进行一个切割的过程。关于信息的收集，可参考文章，自行了解一下：网站分析数据（即用户行为数据）的三种收集方式详解 如果有时间，再整理相应的实操教程出来。

会话切割：用户行为分析的前置工作，相当于一个前置处理过程而已。

2. 学习收获

学完此教程，可以学到平时工作上怎么实现会话切割的一整套流程？理解切割前后的数据是怎样的？更加熟悉Scala API的使用以及其他开发上的技巧等等。

3. 数据源介绍

目前我们已经采集到了三份数据：网站用户点击日志、用户Cookie的标签、网站域名的标签

一、网站用户点击日志（存储在HDFS），格式如下：

#type|server time|cookie|ip|url
pageview|2017-09-04 12:00:00|cookie1|127.0.0.3|https://www.baidu.com
click|2017-09-04 12:00:02|cookie1|127.0.0.3|https://www.baidu.com
pageview|2017-09-04 12:00:01|cookie2|127.0.0.4|https://www.baidu.com
click|2017-09-04 12:00:04|cookie1|127.0.0.3|https://www.baidu.com
pageview|2017-09-04 12:00:02|cookie2|127.0.0.4|http://news.baidu.com
click|2017-09-04 12:00:03|cookie2|127.0.0.4|http://news.baidu.com
pageview|2017-09-04 12:00:04|cookie2|127.0.0.4|http://music.baidu.com/?fr=tieba
pageview|2017-09-04 12:45:01|cookie1|127.0.0.3|https://tieba.baidu.com/index.html
click|2017-09-04 12:45:02|cookie1|127.0.0.3|https://tieba.baidu.com/index.html
click|2017-09-04 12:45:03|cookie1|127.0.0.3|https://tieba.baidu.com/index.html
hhhh|2017-09-04 12:45:03|cookie1|127.0.0.3|https://tieba.baidu.com/index.html
3333ss|2017-09-04 12:45:03|cookie1|127.0.0.3|https://tieba.baidu.com/index.html

解释
此可以通过网站的后台采集到，我们去访问了一次网页时，这是一个行为，我们叫这个行为称为pageview行为；我们点击（click）了网页的某个按钮，也会有一个点击行为；还有在网页中滚动、选定文字等等也是相应的行为……每个行为都会产生相应的信息，至于是什么信息，可以看一下上面的数据里面有什么内容，这些信息都可以被采集到后台。这里为了方便起见，只拿pageview与click这两个行为作为例子，这些行为与Session机制是分不开的。这里的数据的第一行是对应的列含义，除了第一行，其他行的列是以“|”分开。因为这些行为信息会很多，所以我们将其存储到HDFS上。

Session：在计算机中，尤其是在网络应用中，称为“会话控制”。Session对象存储特定用户会话所需的属性及配置信息。这样，当用户在应用程序的Web页之间跳转时，存储在Session对象中的变量将不会丢失，而是在整个用户会话中一直存在下去。当用户请求来自应用程序的 Web页时，如果该用户还没有会话，则Web服务器将自动创建一个 Session对象。当会话过期或被放弃后，服务器将终止该会话。Session 对象最常见的一个用法就是存储用户的首选项。例如，如果用户指明不喜欢查看图形，就可以将该信息存储在Session对象中。有关使用Session 对象的详细信息，请参阅“ASP应用程序”部分的“管理会话”。注意会话状态仅在支持cookie的浏览器中保留。
——来自百度百科

通俗易懂描述：
至于会话，学过浏览器相关知识的人应该都比较清楚什么是会话，会话即session，我们在浏览网站的时候，会对网站发起一次会话，就好像我们用XShell建立一个会话窗口访问我们的服务器一样，那也是一个会话，而且你会发现，如果窗口很久不动它，它会自动断开连接。在我们浏览网站的时候也一样，我们的会话有效时长默认是30min，30min不动，默认断开会话。
这里我们所做的会话切割业务，就是根据我们网站后台记录的时间，每隔30分钟切割成一个会话。
我们现在已经有一个比较粗略概念，即会话有点类似与一种请求，而这个请求，Web服务器可以自动给你创建一个Session对象，里面可以存各种属性，包括pageview、click等等。

二、用户Cookie的标签（存储在HDFS），格式如下：

cookie1|固执
cookie2|有偏见
cookie3|执着
cookie4|执行力很强

解释
每个cookie都打上了对应的类型，作为标签。此过程可以由数据团队，根据机器学习的算法来实现，这里不纠结先，后期有时间会给出教程。因为这些行为信息也会很多，所以我们也将其存储到HDFS上。

Cookie，有时也用其复数形式 Cookies，指某些网站为了辨别用户身份、进行 session 跟踪而储存在用户本地终端上的数据（通常经过加密）。
——来自百度百科

三、网站域名的标签（存储在配置库，如数据库）

"www.baidu.com" -> "level1",
"www.taobao.com" -> "level2", 
"jd.com" -> "level3", 
"youku.com" -> "level4"

解释
我们自定义的给网站分的级别，已经打上了相应的级别标签，当然，也可能有些域名没有分级。这些域名远远没有行为信息那么多，我们可以考虑存储在传统数据库里面。

4. 项目总体流程

其实上面就是一个ETL的过程，最终我们会得到两张表，分别是TrackerLog表与TrackerSession表，也都是存储在HDFS上的，我们这里用的存储方式是Parquet格式。TrackerLog对应我们的原始日志的表，TrackerSession为我们切割后的Session信息表。
我们可以大致用肉眼看一下上面的十行数据，每30min切成成一个会话，我们可以看出来，是有三个会话的，如果我们最后切割出来有三个会话，则表示我们实操结果是对的。
但在此过程中，我们主要是实现的是上图中间的会话切割、生成会话这里边的业务。

5. 最终数据结构

a. 确定表字段

TrackerLog表的字段（暂时可以先忽略里面的数据）：

TrackerSession表字段（暂时可以先忽略里面的数据）：

字段解释
session_id：唯一标识我们的会话
session_server_time：最早访问的那条session创建的时间
landing_url：对应第一个访问的URL，即同属于一个session的最早时间那个
domain：域名

数据解释
观察前面两个表格的数据，表一共有10条，表二的pageview_count与click_count字段的数据加起来正好也是10条的。

b. 定义Schema信息
既然我们是准备用Parquet格式进行存储，那么就要定义好我们的Schema，如：

{"namespace": "com.shaonaiyi.spark.session","type": "record","name": "TrackerLog","fields": [{"name": "log_type", "type": "string"},{"name": "log_server_time", "type": "string"},{"name": "cookie", "type": "string"},{"name": "ip", "type": "string"},{"name": "url", "type": "string"}]
}

{"namespace": "com.shaonaiyi.spark.session","type": "record","name": "TrackerSession","fields": [{"name": "session_id", "type": "string"},{"name": "session_server_time", "type": "string"},{"name": "cookie", "type": "string"},{"name": "cookie_label", "type": "string"},{"name": "ip", "type": "string"},{"name": "landing_url", "type": "string"},{"name": "pageview_count", "type": "int"},{"name": "click_count", "type": "int"},{"name": "domain", "type": "string"},{"name": "domain_label", "type": "string"}]
}

将上面的Schema信息保存成两个文件，分别是：TrackerLog.avsc、TrackerSession.avsc，现在我们的准备工作已经完成了，接下来我们就可以开始构建一个项目了。

0x02 编程实现

1. 构建Maven项目

本人博客里面有很多教程，这里不再重复，相信学到这里的人，已经对这些基础知识相当熟练。
需要注意的是：
1、包名要与前面Schema信息定义的一致，如我的是：com.shaonaiyi.spark.session，当然，你可以继续先操作，不创建包名先。

2. 编码前准备工作

a. 引入依赖及插件（完整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.shaonaiyi</groupId><artifactId>spark-sessioncut</artifactId><version>1.0-SNAPSHOT</version><dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.11</artifactId><version>2.2.0</version><!--<scope>provided</scope>--></dependency><dependency><groupId>org.apache.parquet</groupId><artifactId>parquet-avro</artifactId><version>1.8.1</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.1</version><configuration><source>1.8</source><target>1.8</target><testExcludes><testExclude>/src/test/**</testExclude></testExcludes><encoding>utf-8</encoding></configuration></plugin><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.1.6</version><executions><execution><goals><goal>compile</goal></goals></execution></executions></plugin><plugin><groupId>org.apache.avro</groupId><artifactId>avro-maven-plugin</artifactId><version>1.7.7</version><executions><execution><phase>generate-sources</phase><goals><goal>schema</goal></goals><configuration><sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory><outputDirectory>${project.basedir}/src/main/java</outputDirectory></configuration></execution></executions></plugin><plugin><artifactId>maven-assembly-plugin</artifactId><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id> <!-- this is used for inheritance merges --><phase>package</phase> <!-- 指定在打包节点执行jar包合并操作 --><goals><goal>single</goal></goals></execution></executions></plugin></plugins></build></project>

b. 引入Avro Schema文件
将TrackerLog.avsc、TrackerSession.avsc两个文件上传到项目的avro文件夹（自己新创建）

c. 引入数据源文件并引入Scala SDK
将数据源文件（visit_log.txt、cookie_label.txt）复制到项目的data文件夹（自己新创建），因为在开发的时候，我们一般是先将一部分测试的数据上传到项目来测试，测试完成后，再将路径修改为HDFS上的路径即可，这是开发的流程。创建scala代码源文件夹并引入Scala SDK，下面是目前的项目结构，请参考：

3. 实现源数据的获取

a. 根据Schema文件生成Java对应的类（因为Maven里引入了avro-maven-plugin插件，所以直接点击Maven里的compile即可生成相应的类）

编译完后，可以看到已经生成了相对应的文件，而且自动创建了包：

b. 新建一个工具类RawLogParserUtil：

package com.shaonaiyi.sessionimport com.shaonaiyi.spark.session.TrackerLog/*** @Auther: shaonaiyi@163.com* @Date: 2019/9/12 09:40* @Description: 将每一行原始日志解析成TrackerLog对象*/
object RawLogParserUtil {def parse(line: String): Option[TrackerLog] = {if (line.startsWith("#")) Noneelse {val fields = line.split("\\|")val trackerLog = new TrackerLog()trackerLog.setLogType(fields(0))trackerLog.setLogServerTime(fields(1))trackerLog.setCookie(fields(2))trackerLog.setIp(fields(3))trackerLog.setUrl(fields(4))Some(trackerLog)}}
}

c. 新建一个项目入口类SessionCutETL：

package com.shaonaiyi.sessionimport org.apache.spark.{SparkConf, SparkContext}/*** @Auther: shaonaiyi@163.com* @Date: 2019/9/12 10:09* @Description: 会话切割的程序主入口*/
object SessionCutETL {def main(args: Array[String]): Unit = {var conf = new SparkConf()conf.setAppName("SessionCutETL")conf.setMaster("local")var sc = new SparkContext(conf)val rawRDD: RDD[String] = sc.textFile("data/rawdata/visit_log.txt")rawRDD.collect().foreach(println)sc.stop()}}

执行后，可以看到已经加载到了原始日志的数据：

PS：因为没有配置Hadoop，所以前面会报此错误，可以忽略：

4. 解析日志源数据

a. 继续在SessionCutETL中添加解析日志源数据代码

val parsedLogRDD: RDD[Option[TrackerLog]] = rawRDD.map( line => RawLogParserUtil.parse(line))
parsedLogRDD.collect().foreach(println)

现在去执行的话会报错：

因为Driver与executor通讯的时候需要进行序列号，而Avro插件给我们生成的类中并没有给我们进行序列化，所以还需要给相应的类进行序列号，序列号很简单，只需要在类的后面实现序列化接口 Serializable 即可。
b. 实现TrackerLog类的序列化：

c. 重新执行就可以看到结果了，到此已经实现了我们的日志的解析：

d. 使用flatmap API调整输出格式
查看上面的代码：
val parsedLogRDD: RDD[Option[TrackerLog]] = rawRDD.map( line => RawLogParserUtil.parse(line))
我们返回的类型是RDD[Option[TrackerLog]]，对于返回结果，我们的格式是None、Some()，这种类型，不是我们想要的结果，我们的结果应该更加简洁。所以可以修改此行代码为：
val parsedLogRDD: RDD[TrackerLog] = rawRDD.flatMap( line => RawLogParserUtil.parse(line))
运算结果如下：

此处需要清楚map与flatMap的区别，简而言之：map是一对一的关系，一行就是一个具体的类，所以多少行就有多少个类，然后每个类都切，切了后，这些类还存在；而flatMap则会对所有的行打平成一个类，然后进行切，最后是只有一个类。

5. 日志清洗操作

a. 非法类型的日志说明

仔细关系我们前面的日志，可以看到我们最后有两条数据的类型是hhhh和3333ss，显然不是合法的日志，这里我们需要将其过滤掉，在实际的日志里面脏数据可能各种各样，但处理流程都类似。

hhhh|2017-09-04 12:45:03|cookie1|127.0.0.3|https://tieba.baidu.com/index.html
3333ss|2017-09-04 12:45:03|cookie1|127.0.0.3|https://tieba.baidu.com/index.html

b. 定义合法类型（与main方法同级）
private val logTypeSet = Set("pageview", "click")

c. 修改解析代码，加上过滤条件
val parsedLogRDD: RDD[TrackerLog] = rawRDD.flatMap(RawLogParserUtil.parse(_)) .filter(trackerLog => logTypeSet.contains(trackerLog.getLogType.toString))

d. 重新执行，会发现后面的那两条脏数据已经没有显示出来了

0x03 思路回顾

1、首先，我们需要清楚，我们的大项目其实是网站用户行为分析，而会话切割只是项目里面的小环节，至于网站用户行为分析可以达到什么结果。我们这里边可以简单列举几个，比如说，我们可以通过分析一个人在浏览网站时点击的内容、打开的内容、停留的时间等等，初步判断此人是否对此内容特别感兴趣，得出这个结论之后，我们就可以对此人做定制化的推荐等等，当然这只是一个小小的例子。刚刚我们是从项目开发流程出发，而实际上，我们应该是已经知道我们可以获取到什么信息什么数据，然后，由我们的信息和数据，挖掘我们想要的功能。比如说，你现在有一个产品，需要找到你的商城网站里面、或者其他网站里面某些人对你同类型的产品特别感兴趣，你只需要筛选出这些人，那么我们就可以给这些人推荐你的产品，这些匹配度度高的人就是你的高级客户，你推销产品肯定也会更加顺利。所以呢，我们就要努力朝着这个方向去实现我们的目标，我们目标就是要筛选出这些与你产品匹配度极高的人。
2、所以在此过程中，我们可以根据各种途径，拿到你需要的数据，当然，这些数据也可能是之前已经存在了的，也可能是你确定了目标之后才确定你是否想要的，缺什么数据就想方法去获取，然后整理一下存放到该存放的地方，像这些网页行为，数据量肯定是很大的，点击、打开、滑动等等都是一条数据，每个人都可以产生很多很多条，对于这些量大的数据，我们就可以存储到HDFS上，关于量少的，就可以存到其他地方，不固定，灵活进行存储选型。
3、这里教程没有提供收集数据的教程，如果有浏览过本博客的读者应该会清楚，我们的数据是怎么得到的，比如说Flume就可以实现，还有很多种方式。我们这里已经省略了那些过程，分析完成之后我们就可以进行编码去实现了，因为存储的数据都只是文本文件，而我们要做的，肯定是要先进行数据清洗，把不合法的数据过滤掉，那么根据什么条件进行过滤呢？ 在学习传统数据库的时候，如MySQL，可以用where来确定过滤的条件，比如检索出年龄大于18岁的学生，那么年龄就是一个很关键的因素，而我们的文本文件中，其实是还没有真正去规定好字段的，只是在第一行确定了Schema信息而已，如果不进行字段的切分，那么每一行都要自定义一次，这样显然很麻烦。那么我们就可以统一将文本文件切分成一段一段的，什么字段对应什么意思、对应什么字段。每个字段都属于一行竖行，很显然，我们可以通过构建一个对象来对应上，一个类相当于一行数据，然后一行里面的每个字段，相当于一个对象里面的属性。类->行，属性->字段，所以我们需要在前面定义一个类。
4、我们的Spark作业在进行数据传输的时候，要先进行序列化，所以还要给我们的类进行序列化，Java里面的序列化类是Serializable，搞定之后就可以去获取我们的数据源了，然后清洗、过滤等，过程比较繁多，后期教程会继续完成。

0xFF 总结

1. 其实Spark里面有内置的Kryo序列化接口，性能更好，而且更适合我们当前的应用场景，这个优化点留到项目后期再进行升级。
2. 我们在测试的时候，看到很多启动时的提示信息，很影响我们查看，我们可以通过设置将其屏蔽掉，以方便我们开发，也一样在后期教程再写上。
3. 俗话说得好：“台上一分钟，台下十年功。” 可是写教程不一样，操作半个钟，思路半天，写教程一天。这就是台下半个钟，台上十年功啊！感谢各位读者支持，感谢老汤，关注，评论，加油。
4. 网站用户行为分析项目系列：
   网站用户行为分析项目之会话切割（一）
   网站用户行为分析项目之会话切割（二）
   网站用户行为分析项目之会话切割（三）

---

作者简介：邵奈一
全栈工程师、市场洞察者、专栏编辑
| 公众号 | 微信 | 微博 | CSDN | 简书 |

福利：
邵奈一的技术博客导航
邵奈一 原创不易，如转载请标明出处。

---