一、session聚合统计之重构实现思路
/**
* session聚合统计（统计出访问时长和访问步长，各个区间的session数量占总session数量的比例）
*
* 如果不进行重构，直接来实现，思路：
* 1、actionRDD，映射成的格式
* 2、按sessionid聚合，计算出每个session的访问时长和访问步长，生成一个新的RDD
* 3、遍历新生成的RDD，将每个session的访问时长和访问步长，去更新自定义Accumulator中的对应的值
* 4、使用自定义Accumulator中的统计值，去计算各个区间的比例
* 5、将最后计算出来的结果，写入MySQL对应的表中
*
* 普通实现思路的问题：
* 1、为什么还要用actionRDD，去映射？其实我们之前在session聚合的时候，映射已经做过了。多此一举
* 2、是不是一定要，为了session的聚合这个功能，单独去遍历一遍session？其实没有必要，已经有session数据
* 之前过滤session的时候，其实，就相当于，是在遍历session，那么这里就没有必要再过滤一遍了
*
* 重构实现思路：
* 1、不要去生成任何新的RDD（处理上亿的数据）
* 2、不要去单独遍历一遍session的数据（处理上千万的数据）
* 3、可以在进行session聚合的时候，就直接计算出来每个session的访问时长和访问步长
* 4、在进行过滤的时候，本来就要遍历所有的聚合session信息，此时，就可以在某个session通过筛选条件后
* 将其访问时长和访问步长，累加到自定义的Accumulator上面去
* 5、就是两种截然不同的思考方式，和实现方式，在面对上亿，上千万数据的时候，甚至可以节省时间长达
* 半个小时，或者数个小时
*
* 开发Spark大型复杂项目的一些经验准则：
* 1、尽量少生成RDD
* 2、尽量少对RDD进行算子操作，如果有可能，尽量在一个算子里面，实现多个需要做的功能
* 3、尽量少对RDD进行shuffle算子操作，比如groupByKey、reduceByKey、sortByKey（map、mapToPair）
* shuffle操作，会导致大量的磁盘读写，严重降低性能
* 有shuffle的算子，和没有shuffle的算子，甚至性能，会达到几十分钟，甚至数个小时的差别
* 有shfufle的算子，很容易导致数据倾斜，一旦数据倾斜，简直就是性能杀手（完整的解决方案）
* 4、无论做什么功能，性能第一
* 在传统的J2EE或者.NET后者PHP，软件/系统/网站开发中，我认为是架构和可维护性，可扩展性的重要
* 程度，远远高于了性能，大量的分布式的架构，设计模式，代码的划分，类的划分（高并发网站除外）
*
* 在大数据项目中，比如MapReduce、Hive、Spark、Storm，我认为性能的重要程度，远远大于一些代码
* 的规范，和设计模式，代码的划分，类的划分；大数据，大数据，最重要的，就是性能
* 主要就是因为大数据以及大数据项目的特点，决定了，大数据的程序和项目的速度，都比较慢
* 如果不优先考虑性能的话，会导致一个大数据处理程序运行时间长度数个小时，甚至数十个小时
* 此时，对于用户体验，简直就是一场灾难
*
* 所以，推荐大数据项目，在开发和代码的架构中，优先考虑性能；其次考虑功能代码的划分、解耦合
*
* 我们如果采用第一种实现方案，那么其实就是代码划分（解耦合、可维护）优先，设计优先
* 如果采用第二种方案，那么其实就是性能优先
*
* 讲了这么多，其实大家不要以为我是在岔开话题，大家不要觉得项目的课程，就是单纯的项目本身以及
* 代码coding最重要，其实项目，我觉得，最重要的，除了技术本身和项目经验以外；非常重要的一点，就是
* 积累了，处理各种问题的经验
*
*/Long userid = null;
Date startTime = null;
Date endTime = null;
long stepLength = 0L;
// 遍历session所有的访问行为
while(iterator.hasNext()) {
// 提取每个访问行为的搜索词字段和点击品类字段
Row row = iterator.next();
if(userid == null) {
userid = row.getLong(1);
}
String searchKeyword = null;
Long clickCategoryId = -1L;
searchKeyword = row.getString(5);
try {
clickCategoryId = row.getLong(6);
} catch (Exception e) {
//e.printStackTrace();
//getLong的时候 if (isNullAt(i)) throw new NullPointerException(s"Value at index $i in null")
clickCategoryId = null;
}
// 实际上这里要对数据说明一下
// 并不是每一行访问行为都有searchKeyword何clickCategoryId两个字段的
// 其实，只有搜索行为，是有searchKeyword字段的
// 只有点击品类的行为，是有clickCategoryId字段的
// 所以，任何一行行为数据，都不可能两个字段都有，所以数据是可能出现null值的
if(StringUtils.isNotEmpty(searchKeyword)) {
if(!searchKeywordsBuffer.toString().contains(searchKeyword)) {
searchKeywordsBuffer.append(searchKeyword + ",");
}
}
if(clickCategoryId != null) {
if(!clickCategoryIdsBuffer.toString().contains(
String.valueOf(clickCategoryId))) {
clickCategoryIdsBuffer.append(clickCategoryId + ",");
}
}
//解析出action 发生的时间
Date actionTime = DateUtils.parseTime(row.getString(4));
//在遍历session同时，计算开始时间，结束时间
if(startTime == null) {
startTime = actionTime;
}
if(endTime == null){
endTime = actionTime;
}
if(actionTime.before(startTime)){
startTime = actionTime;
}
if(actionTime.after(endTime)){
endTime = actionTime;
}
//计算访问步长
stepLength++;
}聚合
/**
* 对行为数据按session粒度进行聚合
* @param actionRDD 行为数据RDD
* @return session粒度聚合数据
*/
private static JavaPairRDD aggregateBySession(
SQLContext sqlContext, JavaRDD actionRDD) {
// 现在actionRDD中的元素是Row，一个Row就是一行用户访问行为记录，比如一次点击或者搜索
// 我们现在需要将这个Row映射成的格式
JavaPairRDD sessionid2ActionRDD = actionRDD.mapToPair(
new PairFunction() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(Row row) throws Exception {
//将这个Row映射成的格式
return new Tuple2(row.getString(2), row);
}
});
// 对行为数据按session_id粒度进行分组
JavaPairRDD sessionid2ActionsRDD =
sessionid2ActionRDD.groupByKey();
// 对每一个session分组进行聚合，将session中所有的搜索词和点击品类都聚合起来
// 到此为止，获取的数据格式，如下：
JavaPairRDD userid2PartAggrInfoRDD = sessionid2ActionsRDD.mapToPair(
new PairFunction() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(Tuple2 tuple)
throws Exception {
String sessionid = tuple._1;
Iterator iterator = tuple._2.iterator();
StringBuffer searchKeywordsBuffer = new StringBuffer("");
StringBuffer clickCategoryIdsBuffer = new StringBuffer("");
Long userid = null;
Date startTime = null;
Date endTime = null;
long stepLength = 0L;
// 遍历session所有的访问行为
while(iterator.hasNext()) {
// 提取每个访问行为的搜索词字段和点击品类字段
Row row = iterator.next();
if(userid == null) {
userid = row.getLong(1);
}
String searchKeyword = null;
Long clickCategoryId = -1L;
searchKeyword = row.getString(5);
try {
clickCategoryId = row.getLong(6);
} catch (Exception e) {
//e.printStackTrace();
//getLong的时候 if (isNullAt(i)) throw new NullPointerException(s"Value at index $i in null")
clickCategoryId = null;
}
// 实际上这里要对数据说明一下
// 并不是每一行访问行为都有searchKeyword何clickCategoryId两个字段的
// 其实，只有搜索行为，是有searchKeyword字段的
// 只有点击品类的行为，是有clickCategoryId字段的
// 所以，任何一行行为数据，都不可能两个字段都有，所以数据是可能出现null值的
if(StringUtils.isNotEmpty(searchKeyword)) {
if(!searchKeywordsBuffer.toString().contains(searchKeyword)) {
searchKeywordsBuffer.append(searchKeyword + ",");
}
}
if(clickCategoryId != null) {
if(!clickCategoryIdsBuffer.toString().contains(
String.valueOf(clickCategoryId))) {
clickCategoryIdsBuffer.append(clickCategoryId + ",");
}
}
//解析出action 发生的时间
Date actionTime = DateUtils.parseTime(row.getString(4));
//在遍历session同时，计算开始时间，结束时间
if(startTime == null) {
startTime = actionTime;
}
if(endTime == null){
endTime = actionTime;
}
if(actionTime.before(startTime)){
startTime = actionTime;
}
if(actionTime.after(endTime)){
endTime = actionTime;
}
//计算访问步长
stepLength++;
}
String searchKeywords = StringUtils.trimComma(searchKeywordsBuffer.toString());
String clickCategoryIds = StringUtils.trimComma(clickCategoryIdsBuffer.toString());
//访问ms
long visitLength = (endTime.getTime() - startTime.getTime());
// 大家思考一下
// 我们返回的数据格式，即使
// 但是，这一步聚合完了以后，其实，我们是还需要将每一行数据，跟对应的用户信息进行聚合
// 问题就来了，如果是跟用户信息进行聚合的话，那么key，就不应该是sessionid
// 就应该是userid，才能够跟格式的用户信息进行聚合
// 如果我们这里直接返回，还得再做一次mapToPair算子
// 将RDD映射成的格式，那么就多此一举
// 所以，我们这里其实可以直接，返回的数据格式，就是
// 然后跟用户信息join的时候，将partAggrInfo关联上userInfo
// 然后再直接将返回的Tuple的key设置成sessionid
// 最后的数据格式，还是
// 聚合数据，用什么样的格式进行拼接？
// 我们这里统一定义，使用key=value|key=value
String partAggrInfo = Constants.FIELD_SESSION_ID + "=" + sessionid + "|"
+ Constants.FIELD_SEARCH_KEYWORDS + "=" + searchKeywords + "|"
+ Constants.FIELD_CLICK_CATEGORY_IDS + "=" + clickCategoryIds
+ Constants.FIELD_VISIT_LENGTH +"="+ visitLength+"|"
+ Constants.FIELD_STEP_LENGTH +"=" + stepLength
;
return new Tuple2(userid, partAggrInfo);
}
});
//获取user info
JavaPairRDD userid2InfoRDD = getUserInfo();
// 将session粒度聚合数据，与用户信息进行join
JavaPairRDD userid2FullInfoRDD =
userid2PartAggrInfoRDD.join(userid2InfoRDD);
// 对join起来的数据进行拼接，并且返回格式的数据
JavaPairRDD sessionid2FullAggrInfoRDD = userid2FullInfoRDD.mapToPair(
new PairFunction() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2 call(
Tuple2 tuple)
throws Exception {
String partAggrInfo = tuple._2._1;
Row userInfoRow = tuple._2._2;
String sessionid = StringUtils.getFieldFromConcatString(
partAggrInfo, "\\|", Constants.FIELD_SESSION_ID);
int age = userInfoRow.getInt(3);
String professional = userInfoRow.getString(4);
String city = userInfoRow.getString(5);
String sex = userInfoRow.getString(6);
String fullAggrInfo = partAggrInfo + "|"
+ Constants.FIELD_AGE + "=" + age + "|"
+ Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
+ Constants.FIELD_CITY + "=" + city + "|"
+ Constants.FIELD_SEX + "=" + sex;
return new Tuple2(sessionid, fullAggrInfo);
}
});
return sessionid2FullAggrInfoRDD;
}
//过滤后的结果， 进行访问时长的统计
if(isFilter) { //就是用户需要的数据
// 计算出session的访问时长和访问步长的范围，并进行相应的累加
long visitLength = Long.valueOf(StringUtils.getFieldFromConcatString(
aggrInfo, "\\|", Constants.FIELD_VISIT_LENGTH));
long stepLength = Long.valueOf(StringUtils.getFieldFromConcatString(
aggrInfo, "\\|", Constants.FIELD_STEP_LENGTH));
//统计时长
calculateVisitLength(visitLength);
//统计步长
calculateStepLength(stepLength);
}/**
* 按筛选参数对session粒度聚合数据进行过滤
* @param sessionid2AggrInfoRDD 聚合后的数据
* @param taskParam 过滤的条件
* @return
*/
private static JavaPairRDD filterSessionAndAggr(
JavaPairRDD sessionid2AggrInfoRDD,
JSONObject taskParam, Accumulator sessionAggrStatAccumulator) {
//从数据库中获取过滤的参数
String startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE);
String endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE);
String searchKeys = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS);
System.out.println(startAge+"------"+endAge+"----"+searchKeys);
JavaPairRDD filterRDD = sessionid2AggrInfoRDD.filter(new Function() {
@Override
public Boolean call(Tuple2 tuple) throws Exception {
// 首先，从tuple中，获取聚合数据
//(ce6983270b154f31a7ca67b56f6abcb3,sessionid=ce6983270b154f31a7ca67b56f6abcb3|searchKeywords=温泉|clickCategoryIds=51,79,80|visitLength=2742|stepLength=12|age=2|professional=professional23|city=city25|sex=male)
String aggrInfo = tuple._2;
Boolean isFilter = false;
//1、按照年龄范围进行过滤(startAge, endAge)
int age = Integer.valueOf(
StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_AGE));
if(startAge != null && endAge != null){
if(age>=Integer.valueOf(startAge) && age=1 && visitLength =4 && visitLength =7 && visitLength =10 && visitLength 30 && visitLength 60 && visitLength 180 && visitLength 600 && visitLength 1800) {
sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30m);
}
}
/**
* 计算访问步长范围
* @param stepLength
*/
private void calculateStepLength(long stepLength) {
if(stepLength >= 1 && stepLength = 4 && stepLength = 7 && stepLength = 10 && stepLength 30 && stepLength 60) {
sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_60);
}
}
});
return filterRDD;
}
