本文为《Spark大型电商项目实战》 系列文章之一，主要介绍session信息按照搜索任务进行聚合分析。
用户访问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本事提供的特性。
代码实现
package com.erik.sparkproject.spark;
import java.util.Iterator;
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.erik.sparkproject.conf.ConfigurationManager;
import com.erik.sparkproject.constant.Constants;
import com.erik.sparkproject.dao.ITaskDAO;
import com.erik.sparkproject.domain.Task;
import com.erik.sparkproject.impl.DAOFactory;
import com.erik.sparkproject.test.MockData;
import com.erik.sparkproject.util.*;
import scala.Tuple2;
/**
* @author Erik
*
*/
public class UserVisitSessionAnalyzeSpark {
public static void main(String[] args) {
SparkConf conf =
new SparkConf()
.setAppName(Constants.SPARK_APP_NAME)
.setMaster(
"local");
JavaSparkContext sc =
new JavaSparkContext(conf);
SQLContext sqlContext = getSQLContext(sc.sc());
mockData(sc, sqlContext);
ITaskDAO taskDAO = DAOFactory.getTaskDAO();
long taskid = ParamUtils.getTaskIdFromArgs(args);
Task task = taskDAO.findById(taskid);
JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());
JavaRDD<Row> actionRDD = getActionRDDByDateRange(sqlContext, taskParam);
JavaPairRDD<String, String> sessionid2AggrInfoRDD =
aggregateBySession(sqlContext, actionRDD);
sc.close();
}
/**
* 获取SQLContext
* 如果在本地测试环境的话，那么久生成SQLC哦那text对象
*如果在生产环境运行的话，那么就生成HiveContext对象
* @param sc SparkContext
* @return SQLContext
*/
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);
}
}
/**
* 获取指定日期范围内的用户访问行为数据
* @param sqlContext SQLContext
* @param taskParam 任务参数
* @return 行为数据RDD
*/
private static JavaRDD<Row>
getActionRDDByDateRange(
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();
}
/**
* 对行为数据按sesssion粒度进行聚合
* @param actionRDD 行为数据RDD
* @return session粒度聚合数据
*/
private static JavaPairRDD<String, String>
aggregateBySession(
SQLContext sqlContext, JavaRDD<Row> actionRDD) {
JavaPairRDD<String, Row> sessionid2ActionRDD = actionRDD.mapToPair(
/**
* PairFunction
* 第一个参数，相当于是函数的输入
* 第二个参数和第三个参数，相当于是函数的输出（Tuple），分别是Tuple第一个和第二个值
*/
new PairFunction<Row, String, Row>() {
private static final long serialVersionUID =
1L;
public Tuple2<String, Row>
call(Row row)
throws Exception {
return new Tuple2<String, Row>(row.getString(
2), row);
}
});
JavaPairRDD<String, Iterable<Row>> sessionid2ActionsRDD =
sessionid2ActionRDD.groupByKey();
JavaPairRDD<Long, String> userid2PartAggrInfoRDD = sessionid2ActionsRDD.mapToPair(
new PairFunction<Tuple2<String, Iterable<Row>>, Long, String>() {
private static final long serialVersionUID =
1L;
public Tuple2<Long, String>
call(Tuple2<String, Iterable<Row>> tuple)
throws Exception {
String sessionid = tuple._1;
Iterator<Row> iterator = tuple._2.iterator();
StringBuffer searchKeywordsBuffer =
new StringBuffer(
"");
StringBuffer clickCategoryIdsBuffer =
new StringBuffer(
"");
Long userid =
null;
while(iterator.hasNext()) {
Row row = iterator.next();
if(userid ==
null) {
userid = row.getLong(
1);
}
String searchKeyword = row.getString(
5);
long clickCategoryId = row.getLong(
6);
if(StringUtils.isNotEmpty(searchKeyword)) {
if(!searchKeywordsBuffer.toString().comains(searchKeyword)) {
searchKeywordsBuffer.append(searchKeyword +
",");
}
}
if(!clickCategoryIdsBuffer.toString().contains(
String.valueOf(clickCategoryId))) {
clickCategoryIdsBuffer.append(clickCategoryId +
",");
}
}
String searchKeywords = StringUtils.trimComma(searchKeywordsBuffer.toString());
String clickCategoryIds = StringUtils.trimComma(clickCategoryIdsBuffer.toString());
String partAggrInfo = Constants.FIELD_SESSION_ID +
"=" + sessionid +
"|"
+ Constants.FIELD_SEARCH_KEYWORDS +
"=" + searchKeywords +
"|"
+ Constants.FIELD_CLICK_CATEGORY_IDS +
"=" + clickCategoryIds;
return new Tuple2<Long, String>(userid, partAggrInfo);
}
});
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;
public Tuple2<Long, Row>
call(Row row)
throws Exception {
return new Tuple2<Long, Row>(row.getLong(
0), row);
}
});
JavaPairRDD<Long, Tuple2<String, Row>> userid2FullInfoRDD =
userid2PartAggrInfoRDD.join(userid2InfoRDD);
JavaPairRDD<String, String> sessionid2FullAggrInfoRDD = userid2FullInfoRDD.mapToPair(
new PairFunction<Tuple2<Long, Tuple2<String, Row>>, String, String>() {
private static final long serialVersionUID =
1L;
public Tuple2<String, String>
call(
Tuple2<Long, Tuple2<String, Row>> 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<String, String>(sessionid, fullAggrInfo);
}
});
return sessionid2FullAggrInfoRDD;
}
}
《Spark 大型电商项目实战》源码：https://github.com/Erik-ly/SprakProject
本文为《Spark大型电商项目实战》系列文章之一。 更多文章：Spark大型电商项目实战：http://blog.csdn.net/u012318074/article/category/6744423
