思路：之前模拟创建了两张表：user_visit_action 和 user_info
对于user_visit_action表
1. 通过用户传过来的指定日期范围内，从user_visit_action中查询出指定的用户访问数据 变成 actionRDD
2. 将actionRDDw映射成<Sessionid ,Row>的格式：sessionid2ActionRDD
3. 对行为数据按Session 粒度进行分组 ： sessionid2ActionsRDD
4. /每一个session分组进行聚合，将session中所有的搜索词和点击品类聚合起来 。获取的数据格式： <userid,partAggrInfo(sessionid,searchKeyWords,clickCategoryIds)> 生成userid2PartAggrInfoRDD对于user_info表
查询所有用户数据,并映射成<userid,Row>格式 ：userid2InfoRDD聚合操作
1. 将session粒度聚合数据与用户信息进行join ：userid2FullInfoRDD
2. 对join起来的userid2FullInfoRDD数据进行拼接，并且返回<session,fullAggrInfo>格式数据 ：sessionid2FullAggrInfoRDD
> partAggrInfo = sessionid +searchKeyWords+clickCategoryIds
> fullAggrInfo = partAggrInfo +age +professional +city + seximport java.util.Iterator;
import com.ibeifeng.sparkproject.util.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import com.alibaba.fastjson.JSONObject;
import com.sparkproject.conf.ConfigurationManager;
import com.sparkproject.constant.Constants;
import com.sparkproject.dao.ITaskDAO;
import com.sparkproject.dao.impl.DAOFactory;
import com.sparkproject.domain.Task;
import com.sparkproject.test.MockData;
import com.sparkproject.util.ParamUtils;
import scala.Tuple2;
/**
* 用户访问session 分析spark作业
*
* 接收用户创建的分析任务，用户可能指定的条件如下
* 1. 时间范围：起始日期到结束日期
* 2. 性别：男/女
* 3. 年龄范围
* 4. 职业：多选
* 5. 城市：多选
* 6. 搜索词：多个搜索词，只要某个session中任何一个action搜索过指定的关键词，那么session就符合条件
* 7. 点击品类：多个品类，只要某个session中任何一个action点击过某个品类，那么session就符合条件
*
* spark作业接受用户创建的任务方法：j2ee在接受用户创建任务请求后，会将其插入mysql的task表中，任务参数以json格式封装在task_param字段中
* 接着，j2ee平台会执行spark-submit shell脚本，并将taskid作为参数传递给spark-submit shell脚本
* spark-submit shell脚本在执行时是可以接收参数的，并且将参数传递给spark作业的main函数。参数就封装在main函数的args数组中
* （这是spark本身提供的特性）
* @author chenc
*
*/
public class UserVisitSessionAnalyzeSpark {
public static void main(String[] args) {
//构建spark上下文
SparkConf conf = new SparkConf()
.setAppName(Constants.SPARK_APP_NAME_SESSION)
.setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);
//从JavaSparkContext取出它自己对应的那个SparkContext
SQLContext sqlContext = getSQLContext(sc.sc());
//生成模拟测试数据
mockData(sc,sqlContext);
//创建需要使用的DAO组件
//用接口类型定义taskDAO
ITaskDAO taskDAO = DAOFactory.getTaskDAO();
//如果要根据用户在创建任务时指定的参数来进行数据过滤和筛选，
//首先得查询出指定的任务,并获取任务的查询参数
//需要先获取任务的id(从封装好的ParamUtils中的getTaskIdFromArgs方法获取）
//（spark-submit shell脚本在执行时是可以接收参数的，并且将参数传递给spark作业的main函数。参数就封装在main函数的args数组中）
Long taskid = ParamUtils.getTaskIdFromArgs(args);
//通过基于JDBC的TaskDAO组件，查询对应的task信息
Task task = taskDAO.findById(taskid);
//将JSON串转换为jsonObject-->拿到了用户的创建参数
JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());
//如果要进行session粒度的数据聚合，首先要从user_visit_action表中，查询出来指定日期范围内的行为数据
JavaRDD<Row> actionRDD = getActionRDDByDataRange(sqlContext, taskParam);
// 首先将行为数据按照session_id 进行groupByKey进行分组
// 此时的数据的粒度是session粒度。然后将session粒度的数据与用户信息数据进行join.就可以获取session粒度的数据
// 同时数据里还包含了session对应的user的信息
JavaPairRDD<String, String> sessionid2AggrInfoRDD = aggregateBySession(sqlContext, actionRDD);
//关闭
sc.close();
}
/**
* 获取SQLContext
* 如果是本地测试环境，生成SQLContext对象
* 如果是生产环境运行，生成HiveContext对象
* @param sc SparkContext
* @return SparkContext
*/
private static SQLContext getSQLContext(SparkContext sc) {
boolean local = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
if (local) {
return new SQLContext(sc);
} else {
return new HiveContext(sc);
}
}
/**
* 生成模拟数据（只有是本地模式才会生成）
* @param sc
* @param sqlContext
*/
private static void mockData(JavaSparkContext sc,SQLContext sqlContext) {
boolean local = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
if(local) {
MockData.mock(sc, sqlContext);
}
}
/**
* 通过用户传过来的指定日期范围内，从user_visit_action中查询出指定的用户访问数据
* @param sqlContext
* @param taskParam 任务参数
* @return 行为数据RDD
*/
private static JavaRDD<Row> getActionRDDByDataRange(SQLContext sqlContext,JSONObject taskParam){
String startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE);
String endDate = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE);
String sql = "select * "
+"from user_visit_action "
+ "where date>='" + startDate + "' "
+ "and date<='" + endDate + "'";
DataFrame actionDF = sqlContext.sql(sql);
return actionDF.javaRDD();
}
/**
* 对行为数据按session粒度进行聚合
* @param actionRDD 行为数据RDD
* @return session粒度聚合数据
*/
private static JavaPairRDD<String, String> aggregateBySession(
SQLContext sqlContext,JavaRDD<Row> actionRDD){
//现在action中的元素是Row，一个Row是一行用户访问行为记录：比如一次点击或者搜索
//现在我们需要将这个Row映射成<Sessionid ,Row>的格式
JavaPairRDD<String, Row> sessionid2ActionRDD = actionRDD.mapToPair(
/**
* PairFunction
* 第一个参数：相当于是函数的输入
* 第二个和第三个参数：相当于函数的输出（Tuple），分别是Tuple第一个和第二个值
*/
new PairFunction<Row, String, Row>() {
private static final long SerialVersionUID= 1L;
@Override
public Tuple2<String, Row> call(Row row) throws Exception {
//第2个是sesion_id
return new Tuple2<String, Row>(row.getString(2),row);
}
});
//对行为数据按Session 粒度进行分组
JavaPairRDD<String, Iterable<Row>> sessionid2ActionsRDD = sessionid2ActionRDD.groupByKey();
//对每一个session分组进行聚合，将session中所有的搜索词和点击品类聚合起来
//获取的数据格式： <userid,partAggrInfo(sessionid,searchKeyWords,clickCategoryIds)>
JavaPairRDD<Long, String> userid2PartAggrInfoRDD = sessionid2ActionsRDD.mapToPair(
new PairFunction<Tuple2<String,Iterable<Row>>, Long , String>() {
private static final long SerialVersionUID= 1L;
@Override
public Tuple2<Long, String> call(Tuple2<String, Iterable<Row>> tuple) throws Exception {
String sessionid=tuple._1;
Iterator<Row> iterator = tuple._2.iterator();
//在遍历的时候，要把它搜索过的关键词和它点击过的品类ID给拼起来
StringBuffer searchKeywordsBuffer = new StringBuffer("");
StringBuffer clickCategoryIdsBuffer = new StringBuffer("");
Long userid = null;
//遍历session所有的访问行为
while(iterator.hasNext()) {
//提取每个访问行为的搜索词字段和点击品类字段
Row row = iterator.next();
//获取userid
if (userid == null) {
userid = row.getLong(1);
}
//search_keyword:5
String searchKeyword = row.getString(5);
//click_category_id:6
Long clickCategoryId = row.getLong(6);
//对数据说明：并非每一行访问行为都有searchKeyword和clickCategoryId 。 只有搜索行为有searchKeyword
//只有点击品类行为有clickCategoryId。所以数据可能出现null
//
//我们决定是否要将搜索词或者点击品类id拼接到字符串中去，首先要满足：
//1. 不能是null值
//2. 之前的字符串中还没有搜索词或者点击品类id
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+",");
}
}
}
//聚合
String searchKeyWords = StringUtils.trimComma(searchKeywordsBuffer.toString());
String clickCategoryIds = StringUtils.trimComma(clickCategoryIdsBuffer.toString());
//返回的数据格式<sessionid,partAggrInfo>
//但是这一步聚合完之后，我们还需要将每一行数据跟对应得用户信息进行聚合
//问题在：若跟用户信息进行聚合，key就不应该是sessionid,而应该是userid才可以
//跟<userid,Row>格式的用户信息进行聚合
//若这里直接返回<sessionid,partAggrInfo>还需要再做一次mapToPair算子将RDD映射为
//<userid,partAggrInfo>，那么会多此一举
//所以此处其实可以直接返回的数据格式就是<userid,partAggrInfo>
//然后和用户信息join的时候，将partAggrInfo关联上userInfo。然后再直接将返回的Tuple的key设置成sessionid
//最后的数据格式还是<sessionid,fullAggrInfo>
//聚合数据用什么格式拼接？
//使用 key=value|key=value
String partAggrInfo = Constants.FIELD_SESSION_ID + "=" + sessionid + "|"
+ Constants.FIELD_SEARCH_KEYWORDS +"=" + searchKeyWords + "|"
+ Constants.FIELD_CLICK_GATEGORY_IDS + "=" + clickCategoryIds;
//返回：session 对应得userid 包括这个session部分聚合数据
return new Tuple2<Long ,String>(userid,partAggrInfo);
}
});
//查询所有用户数据,并映射成<userid,Row>格式
String sql = " select * from user_info";
JavaRDD<Row> userInfoRDD = sqlContext.sql(sql).javaRDD();
JavaPairRDD<Long, Row> userid2InfoRDD = userInfoRDD.mapToPair(new PairFunction<Row, Long, Row>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2<Long, Row> call(Row row) throws Exception {
return new Tuple2<Long,Row>(row.getLong(0),row);
}
});
//将session粒度聚合数据与用户信息进行join
JavaPairRDD<Long, Tuple2<String, Row>> userid2FullInfoRDD = userid2PartAggrInfoRDD.join(userid2InfoRDD);
//对join起来的数据进行拼接，并且返回<session,fullAggrInfo>格式数据
JavaPairRDD<String, String> sessionid2FullAggrInfoRDD = userid2FullInfoRDD.mapToPair(new PairFunction<Tuple2<Long,Tuple2<String,Row>>, String, String>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2<String, String> call(Tuple2<Long, Tuple2<String, Row>> tuple) throws Exception {
String partAggrInfo = tuple._2._1;
Row userInfoRow = tuple._2._2;
//获取session_id
String sessionid = StringUtils.getFieldFromConcatString(partAggrInfo, "\\|", Constants.FIELD_SESSION_ID);
//从user表中取出age(3)/professional(4)/city(5)/sex(6)
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<String,String>(sessionid,fullAggrInfo);
}
}) ;
return sessionid2FullAggrInfoRDD;
}
}
之前已经获取了通过条件过滤的session，要实现：访问时长在0~3S内的session的数量，占总session数量的比例；4~6s。。。
访问时长：把session的最后一个action的时间减去第一个action的时间
访问步长：session的action的数量（访问页面次数）
采用的方式思路：自定义一个Accumulator，实现较为复杂的计算逻辑，一个Accumulator维护了所有范围区间的数量的统计逻辑。
1. 低耦合。如果说，session数量计算逻辑要改变，那么不用变更session遍历相关的代码，只要维护一个Accumulator里面的代码即可。
2. 如果计算逻辑后期变更或者增加了几个范围，那么也很方便，不用多加几个Accumulator、去修改大量代码。只要维护一个Accumulator里面的代码即可
使用自定义accumulator，可以任意实现自己的复杂分布式计算的逻辑。如果task是分布式的，要进行复杂计算逻辑那么是很难实现的，（可能需要借助redis，维护中间状态，借助于zookeeper实现分布式锁）。但是使用自定义accumulator，可以更方便进行中间状态的维护，不用担心并发和锁的问题。
重构实现思路和重构session聚合
/**
* session 聚合统计（统计出访问时长和访问不常，各个区间session数量占据总session数量的比例）
* 若不进行重构直接实现：
* 1. actionRDD 映射成<sessionid,Row>格式.
* 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上
*
* 第一种方案其实是代码划分（解耦合、可维护）优先，设计优先。第二种方案是性能优先
重构过滤进行统计
计算统计结果并且存入MYSQL
/**
* 计算各session范围占比，并写入MySQL
* @param value
*/
private static void calculateAndPersistAggrStat(String value, long taskid) {
// 从Accumulator统计串中获取值
long session_count = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.SESSION_COUNT));
long visit_length_1s_3s = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.TIME_PERIOD_1s_3s));
long visit_length_4s_6s = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.TIME_PERIOD_4s_6s));
long visit_length_7s_9s = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.TIME_PERIOD_7s_9s));
long visit_length_10s_30s = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.TIME_PERIOD_10s_30s));
long visit_length_30s_60s = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.TIME_PERIOD_30s_60s));
long visit_length_1m_3m = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.TIME_PERIOD_1m_3m));
long visit_length_3m_10m = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.TIME_PERIOD_3m_10m));
long visit_length_10m_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.TIME_PERIOD_10m_30m));
long visit_length_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.TIME_PERIOD_30m));
long step_length_1_3 = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.STEP_PERIOD_1_3));
long step_length_4_6 = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.STEP_PERIOD_4_6));
long step_length_7_9 = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.STEP_PERIOD_7_9));
long step_length_10_30 = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.STEP_PERIOD_10_30));
long step_length_30_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.STEP_PERIOD_30_60));
long step_length_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
value, "\\|", Constants.STEP_PERIOD_60));
// 计算各个访问时长和访问步长的范围
double visit_length_1s_3s_ratio = NumberUtils.formatDouble(
(double)visit_length_1s_3s / (double)session_count, 2);
double visit_length_4s_6s_ratio = NumberUtils.formatDouble(
(double)visit_length_4s_6s / (double)session_count, 2);
double visit_length_7s_9s_ratio = NumberUtils.formatDouble(
(double)visit_length_7s_9s / (double)session_count, 2);
double visit_length_10s_30s_ratio = NumberUtils.formatDouble(
(double)visit_length_10s_30s / (double)session_count, 2);
double visit_length_30s_60s_ratio = NumberUtils.formatDouble(
(double)visit_length_30s_60s / (double)session_count, 2);
double visit_length_1m_3m_ratio = NumberUtils.formatDouble(
(double)visit_length_1m_3m / (double)session_count, 2);
double visit_length_3m_10m_ratio = NumberUtils.formatDouble(
(double)visit_length_3m_10m / (double)session_count, 2);
double visit_length_10m_30m_ratio = NumberUtils.formatDouble(
(double)visit_length_10m_30m / (double)session_count, 2);
double visit_length_30m_ratio = NumberUtils.formatDouble(
(double)visit_length_30m / (double)session_count, 2);
double step_length_1_3_ratio = NumberUtils.formatDouble(
(double)step_length_1_3 / (double)session_count, 2);
double step_length_4_6_ratio = NumberUtils.formatDouble(
(double)step_length_4_6 / (double)session_count, 2);
double step_length_7_9_ratio = NumberUtils.formatDouble(
(double)step_length_7_9 / (double)session_count, 2);
double step_length_10_30_ratio = NumberUtils.formatDouble(
(double)step_length_10_30 / (double)session_count, 2);
double step_length_30_60_ratio = NumberUtils.formatDouble(
(double)step_length_30_60 / (double)session_count, 2);
double step_length_60_ratio = NumberUtils.formatDouble(
(double)step_length_60 / (double)session_count, 2);
// 将统计结果封装为Domain对象
SessionAggrStat sessionAggrStat = new SessionAggrStat();
sessionAggrStat.setTaskid(taskid);
sessionAggrStat.setSession_count(session_count);
sessionAggrStat.setVisit_length_1s_3s_ratio(visit_length_1s_3s_ratio);
sessionAggrStat.setVisit_length_4s_6s_ratio(visit_length_4s_6s_ratio);
sessionAggrStat.setVisit_length_7s_9s_ratio(visit_length_7s_9s_ratio);
sessionAggrStat.setVisit_length_10s_30s_ratio(visit_length_10s_30s_ratio);
sessionAggrStat.setVisit_length_30s_60s_ratio(visit_length_30s_60s_ratio);
sessionAggrStat.setVisit_length_1m_3m_ratio(visit_length_1m_3m_ratio);
sessionAggrStat.setVisit_length_3m_10m_ratio(visit_length_3m_10m_ratio);
sessionAggrStat.setVisit_length_10m_30m_ratio(visit_length_10m_30m_ratio);
sessionAggrStat.setVisit_length_30m_ratio(visit_length_30m_ratio);
sessionAggrStat.setStep_length_1_3_ratio(step_length_1_3_ratio);
sessionAggrStat.setStep_length_4_6_ratio(step_length_4_6_ratio);
sessionAggrStat.setStep_length_7_9_ratio(step_length_7_9_ratio);
sessionAggrStat.setStep_length_10_30_ratio(step_length_10_30_ratio);
sessionAggrStat.setStep_length_30_60_ratio(step_length_30_60_ratio);
sessionAggrStat.setStep_length_60_ratio(step_length_60_ratio);
// 调用对应的DAO插入统计结果
ISessionAggrStatDAO sessionAggrStatDAO = DAOFactory.getSessionAggrStatDAO();
sessionAggrStatDAO.insert(sessionAggrStat); 结果界面
应该在本地mysql中的session_aggr_stat表中存在如下数据即完成。
