基于Spark的电商用户行为分析大数据平台项目源码+数据集.rar

package cn.edu.hust.session; import cn.edu.hust.conf.ConfigurationManager; import cn.edu.hust.constant.Constants; import cn.edu.hust.dao.TaskDao; import cn.edu.hust.dao.factory.DaoFactory; import cn.edu.hust.domain.*; import cn.edu.hust.mockData.MockData; import cn.edu.hust.util.*; import com.alibaba.fastjson.JSONObject; import org.apache.spark.Accumulator; 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.*; 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 org.apache.spark.storage.StorageLevel; import scala.Tuple2; import java.util.*; /** * 用户可以查询的范围包含 * 1。用户的职业 * 2。用户的性别 * 3。用户城市 * 4。用户年龄 * 5。获取搜索词 * 6。获取点击品类 */ public class UserVisitAnalyze { public static void main(String[] args) { args=new String[]{"1"}; /** * 构建spark上下文 */ SparkConf conf=new SparkConf().setAppName(Constants.APP_NAME_SESSION).setMaster("local[3]"); JavaSparkContext context=new JavaSparkContext(conf); SQLContext sc=getSQLContext(context.sc()); //生成模拟数据 mock(context,sc); //拿到相应的Dao组建 TaskDao dao= DaoFactory.getTaskDao(); //从外部传入的参数获取任务的id Long taskId=ParamUtils.getTaskIdFromArgs(args); //从数据库中查询出相应的task Task task=dao.findTaskById(taskId); JSONObject jsonObject=JSONObject.parseObject(task.getTaskParam()); //获取指定范围内的Sesssion JavaRDD<Row> sessionRangeDate=getActionRDD(sc,jsonObject); //这里增加一个新的方法，主要是映射 JavaPairRDD<String,Row> sessionInfoPairRDD=getSessonInfoPairRDD(sessionRangeDate); //重复用到的RDD进行持久化 sessionInfoPairRDD.persist(StorageLevel.DISK_ONLY()); //上面的两个RDD是 //按照Sesson进行聚合 JavaPairRDD<String,String> sesssionAggregateInfoRDD=aggregateBySessionId(sc,sessionInfoPairRDD); //通过条件对RDD进行筛选 // 重构，同时统计 Accumulator<String> sessionAggrStatAccumulator=context.accumulator("",new SessionAggrStatAccumulator()); //在进行accumulator之前，需要aciton动作，不然会为空 JavaPairRDD<String,String> filteredSessionRDD=filterSessionAndAggrStat(sesssionAggregateInfoRDD,jsonObject,sessionAggrStatAccumulator); //重复用到的RDD进行持久化 filteredSessionRDD.persist(StorageLevel.DISK_ONLY()); //获取符合过滤条件的全信息公共RDD JavaPairRDD<String, Row> commonFullClickInfoRDD=getFilterFullInfoRDD(filteredSessionRDD,sessionInfoPairRDD); //重复用到的RDD进行持久化 commonFullClickInfoRDD.persist(StorageLevel.DISK_ONLY()); //session聚合统计，统计出访问时长和访问步长的各个区间所占的比例 /** * 重构实现的思路： * 1。不要去生成任何的新RDD * 2。不要去单独遍历一遍sesion的数据 * 3。可以在聚合数据的时候可以直接计算session的访问时长和访问步长 * 4。在以前的聚合操作中，可以在以前的基础上进行计算加上自己实现的Accumulator来进行一次性解决 * 开发Spark的经验准则 * 1。尽量少生成RDD * 2。尽量少对RDD进行蒜子操作，如果可能，尽量在一个算子里面，实现多个需求功能 * 3。尽量少对RDD进行shuffle算子操作，比如groupBykey、reduceBykey、sortByKey * shuffle操作，会导致大量的磁盘读写，严重降低性能 * 有shuffle的算子，和没有shuffle的算子，甚至性能相差极大 * 有shuffle的算子，很容易造成性能倾斜，一旦数据倾斜，简直就是性能杀手 * 4。无论做什么功能，性能第一 * 在大数据项目中，性能最重要。主要是大数据以及大数据项目的特点，决定了大数据的程序和项目速度，都比较满 * 如果不考虑性能的话，就会导致一个大数据处理程序运行长达数个小时，甚至是数个小时，对用户的体验，简直是 * 一场灾难。 */ /** * 使用CountByKey算子实现随机抽取功能 */ randomExtractSession(taskId,filteredSessionRDD,sessionInfoPairRDD); //在使用Accumulutor之前，需要使用Action算子，否则获取的值为空，这里随机计算 //filteredSessionRDD.count(); //计算各个session占比,并写入MySQL calculateAndPersist(sessionAggrStatAccumulator.value(),taskId); //获取热门品类数据Top10 List<Tuple2<CategorySortKey,String>> top10CategoryIds=getTop10Category(taskId,commonFullClickInfoRDD); //获取热门每一个品类点击Top10session getTop10Session(context,taskId,sessionInfoPairRDD,top10CategoryIds); //关闭spark上下文 context.close(); } /** * 用于判断是否是生产环境 * @param sc * @return */ public static SQLContext getSQLContext(SparkContext sc) { boolean local= ConfigurationManager.getBoolean(Constants.SPARK_LOCAL); if(local) { return new SQLContext(sc); } return new HiveContext(sc); } private static void mock(JavaSparkContext context,SQLContext sc) { boolean local= ConfigurationManager.getBoolean(Constants.SPARK_LOCAL); if(local) { MockData.mock(context,sc); } } /** * 获取指定日期范围内的数据 * @param sc * @param taskParam * @return */ private static JavaRDD<Row> getActionRDD(SQLContext sc, JSONObject taskParam) { String startTime=ParamUtils.getParam(taskParam,Constants.PARAM_STARTTIME); String endTime=ParamUtils.getParam(taskParam,Constants.PARAM_ENDTIME); String sql="select *from user_visit_action where date>='"+startTime+"' and date<='"+endTime+"'"; DataFrame df=sc.sql(sql); return df.javaRDD(); } /** * 将数据进行映射成为Pair，键为SessionId，Value为Row * @param sessionRangeDate * @return */ private static JavaPairRDD<String,Row> getSessonInfoPairRDD(JavaRDD<Row> sessionRangeDate) { return sessionRangeDate.mapToPair(new PairFunction<Row, String, Row>() { @Override public Tuple2<String, Row> call(Row row) throws Exception { return new Tuple2<String, Row>(row.getString(2),row); } }); } /** * session粒度的聚合 * @param sc * @param sessionInfoPairRDD * @return */ private static JavaPairRDD<String,String> aggregateBySessionId(SQLContext sc, JavaPairRDD<String, Row> sessionInfoPairRDD) { /** * 先将数据映射成map格式 */ /** * 代码重构 JavaPairRDD<String,Row> sessionActionPair=sessionRangeDate.mapToPair(new PairFunction<Row, String,Row>() { @Override public Tuple2<String, Row> call(Row row) throws Exception { return new Tuple2<String, Row>(row.getString(2),row); } });*/ /**