本项目主要用于互联网电商企业中，使用Spark技术开发的大数据统计分析平台，对电商网站的各种用户行为（访问行为、购物行为、广告点击行为等）进行复杂的分析。用统计分析出来的数据，辅助公司中的PM（产品经理）、数据分析师以及管理人员分析现有产品的情况，并根据用户行为分析结果持续改进产品的设计，以及调整公司的战略和业务。最终达到用大数据技术来帮助提升公司的业绩、营业额以及市场占有率的目标。

本项目使用了Spark技术生态栈中最常用的三个技术框架，Spark Core、Spark SQL和Spark Streaming，进行离线计算和实时计算业务模块的开发。实现了包括用户访问session分析、页面单跳转化率统计、热门商品离线统计、广告流量实时统计4个业务模块。

在项目中，重点介绍符合实际企业项目中的性能调优、troubleshooting以及数据倾斜等知识和技术，同时以企业级大数据项目开发流程贯穿每个业务模块的讲解，涵盖了项目开发全流程，包括需求分析、方案设计、数据设计、编码实现、测试以及性能调优等环节，全面还原真实大数据项目的开发流程。

模块介绍

用户访问session分析

该模块主要是对用户访问session进行统计分析，包括session的聚合指标计算、按时间比例随机抽取session、获取每天点击、下单和购买排名前10的品类、并获取top10品类的点击量排名前10的session。该模块可以让产品经理、数据分析师以及企业管理层形象地看到各种条件下的具体用户行为以及统计指标，从而对公司的产品设计以及业务发展战略做出调整。主要使用Spark Core实现。

页面单跳转化率统计

该模块主要是计算关键页面之间的单步跳转转化率，涉及到页面切片算法以及页面流匹配算法。该模块可以让产品经理、数据分析师以及企业管理层看到各个关键页面之间的转化率，从而对网页布局，进行更好的优化设计。主要使用Spark Core实现。

热门商品离线统计

该模块主要实现每天统计出各个区域的top3热门商品。然后使用Oozie进行离线统计任务的定时调度；使用Zeppeline进行数据可视化的报表展示。该模块可以让企业管理层看到公司售卖的商品的整体情况，从而对公司的商品相关的战略进行调整。主要使用Spark SQL实现。

参考： https://github.com/Erik-ly/SprakProject
参考：https://blog.csdn.net/u012318074/category_6744423.html

用户访问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) {
        //构建spark上下文
        //首先在Constants.java中设置spark作业相关的常量
        //String SPARK_APP_NAME = "UserVisitSessionAnalyzeSpark";
        //保存Constants.java配置
        SparkConf conf = new SparkConf()
                .setAppName(Constants.SPARK_APP_NAME)
                .setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = getSQLContext(sc.sc());
        //生成模拟测试数据
        mockData(sc, sqlContext);
        //创建需要使用的DAO组件
        ITaskDAO taskDAO = DAOFactory.getTaskDAO();
        //那么就首先得查询出来指定的任务，并获取任务的查询参数
        long taskid = ParamUtils.getTaskIdFromArgs(args);
        Task task = taskDAO.findById(taskid);
        JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());
        //如果要进行session粒度的数据聚合，
        //首先要从user_visit_action表中，查询出来指定日期范围内的数据
        JavaRDD<Row> actionRDD = getActionRDDByDateRange(sqlContext, taskParam);
        //聚合
        //首先，可以将行为数据按照session_id进行groupByKey分组
        //此时的数据粒度就是session粒度了，然后可以将session粒度的数据与用户信息数据惊醒join
        //然后就可以获取到session粒度的数据，同时数据里面还包含了session对应的user信息
        JavaPairRDD<String, String> sessionid2AggrInfoRDD = 
                aggregateBySession(sqlContext, actionRDD);
        //关闭spark上下文
        sc.close();
    }
    /**
     * 获取SQLContext
     * 如果在本地测试环境的话，那么久生成SQLC哦那text对象
     *如果在生产环境运行的话，那么就生成HiveContext对象
     * @param sc SparkContext
     * @return SQLContext
     */
    private static SQLContext getSQLContext(SparkContext sc) {
        //在my.properties中配置
        //spark.local=true（打包之前改为flase）
        //在ConfigurationManager.java中添加
        //public static Boolean getBoolean(String key) {
        //  String value = getProperty(key);
        //  try {
        //      return Boolean.valueOf(value);
        //  } catch (Exception e) {
        //      e.printStackTrace();
        //  }
        //  return false;   
        //}
        //在Contants.java中添加
        //String SPARK_LOCAL = "spark.local";
        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) {
        //先在Constants.java中添加任务相关的常量
        //String PARAM_START_DATE = "startDate";
        //String PARAM_END_DATE = "endDate";
        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) {
        //现在actionRDD中的元素是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;
                    public Tuple2<String, Row> call(Row row) throws Exception {
                        //按照MockData.java中字段顺序获取
                        //此时需要拿到session_id，序号是2
                        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;
                    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;
                        //遍历session所有的访问行为
                        while(iterator.hasNext()) {
                            //提取每个 访问行为的搜索词字段和点击品类字段
                            Row row = iterator.next();
                            if(userid == null) {
                                userid = row.getLong(1);
                            }
                            String searchKeyword = row.getString(5);
                            long clickCategoryId = row.getLong(6);
                            //实际上这里要对数据说明一下
                            //并不是每一行访问行为都有searchKeyword和clickCategoryId两个字段的
                            //其实，只有搜索行为是有searchKeyword字段的
                            //只有点击品类的行为是有clickCaregoryId字段的
                            //所以，任何一行行为数据，都不可能两个字段都有，所以数据是可能出现null值的
                            //所以是否将搜索词点击品类id拼接到字符串中去
                            //首先要满足不能是null值
                            //其次，之前的字符串中还没有搜索词或者点击品类id
                            if(StringUtils.isNotEmpty(searchKeyword)) {
                                  if(!searchKeywordsBuffer.toString().comains(searchKeyword)) {
                                    searchKeywordsBuffer.append(searchKeyword + ",");
                                }
                            }
                            if(!clickCategoryIdsBuffer.toString().contains(
                                    String.valueOf(clickCategoryId))) {
                                clickCategoryIdsBuffer.append(clickCategoryId + ",");
                            }
                        }
                        //StringUtils引入的包是import com.erik.sparkproject.util.trimComma;
                        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>
                        //然后在直接将返回的Tuple的key设置成sessionid
                        //最后的数据格式，还是<sessionid,fullAggrInfo>
                        //聚合数据，用什么样的格式进行拼接？
                        //我们这里统一定义，使用key=value|key=vale
                        //在Constants.java中定义spark作业相关的常量
                        //String FIELD_SESSION_ID = "sessionid";
                        //String FIELD_SEARCH_KEYWORDS = "searchKeywords";
                        //String FIELD_CLICK_CATEGORY_IDS = "clickCategoryIds";
                        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);
                    }
                });
        //将session粒度聚合数据，与用户信息进行join
        JavaPairRDD<Long, Tuple2<String, Row>> userid2FullInfoRDD = 
                userid2PartAggrInfoRDD.join(userid2InfoRDD);
        //对join起来的数据进行拼接，并且返回<sessionid,fullAggrInfo>格式的数据
        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);
                        //在Constants.java中添加以下常量
                        //String FIELD_AGE = "age";
                        //String FIELD_PROFESSIONAL = "professional";
                        //String FIELD_CITY = "city";
                        //String FIELD_SEX = "sex";
                        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粒度聚合数据进行过滤

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.Function;
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) {
        args = new String[]{"2"}; 
        //构建spark上下文
        //首先在Constants.java中设置spark作业相关的常量
        //String SPARK_APP_NAME = "UserVisitSessionAnalyzeSpark";
        //保存Constants.java配置
        SparkConf conf = new SparkConf()
                .setAppName(Constants.SPARK_APP_NAME)
                .setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = getSQLContext(sc.sc());
        //生成模拟测试数据
        mockData(sc, sqlContext);
        //创建需要使用的DAO组件
        ITaskDAO taskDAO = DAOFactory.getTaskDAO();
        //那么就首先得查询出来指定的任务，并获取任务的查询参数
        long taskid = ParamUtils.getTaskIdFromArgs(args);
        Task task = taskDAO.findById(taskid);
        JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());
        //如果要进行session粒度的数据聚合，
        //首先要从user_visit_action表中，查询出来指定日期范围内的数据
        JavaRDD<Row> actionRDD = getActionRDDByDateRange(sqlContext, taskParam);
        //聚合
        //首先，可以将行为数据按照session_id进行groupByKey分组
        //此时的数据粒度就是session粒度了，然后可以将session粒度的数据与用户信息数据惊醒join
        //然后就可以获取到session粒度的数据，同时数据里面还包含了session对应的user信息
        //到这里为止，获取的数据是<sessionid,(sessionid,searchKeywords,
        //clickCategoryIds,age,professional,city,sex)>
        JavaPairRDD<String, String> sessionid2AggrInfoRDD = 
                aggregateBySession(sqlContext, actionRDD);
        //接着，就要针对session粒度的聚合数据，按照使用者指定的筛选参数进行数据过滤
        //相当于我们自己编写的算子，是要访问外面的任务参数对象的
        //匿名内部类（算子函数），访问外部对象，是要给外部对象使用final修饰的
        JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD = 
                filterSession(sessionid2AggrInfoRDD, taskParam);
        //关闭spark上下文
        sc.close();
    }
    /**
     * 获取SQLContext
     * 如果在本地测试环境的话，那么久生成SQLC哦那text对象
     *如果在生产环境运行的话，那么就生成HiveContext对象
     * @param sc SparkContext
     * @return SQLContext
     */
    private static SQLContext getSQLContext(SparkContext sc) {
        //在my.properties中配置
        //spark.local=true（打包之前改为flase）
        //在ConfigurationManager.java中添加
        //public static Boolean getBoolean(String key) {
        //  String value = getProperty(key);
        //  try {
        //      return Boolean.valueOf(value);
        //  } catch (Exception e) {
        //      e.printStackTrace();
        //  }
        //  return false;   
        //}
        //在Contants.java中添加
        //String SPARK_LOCAL = "spark.local";
        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) {
        //先在Constants.java中添加任务相关的常量
        //String PARAM_START_DATE = "startDate";
        //String PARAM_END_DATE = "endDate";
        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) {
        //现在actionRDD中的元素是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;
                    public Tuple2<String, Row> call(Row row) throws Exception {
                        //按照MockData.java中字段顺序获取
                        //此时需要拿到session_id，序号是2
                        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;
                    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;
                        //遍历session所有的访问行为
                        while(iterator.hasNext()) {
                            //提取每个 访问行为的搜索词字段和点击品类字段
                            Row row = iterator.next();
                            if(userid == null) {
                                userid = row.getLong(1);
                            }
                            String searchKeyword = row.getString(5);
                            Long clickCategoryId = row.getLong(6);
                            //实际上这里要对数据说明一下
                            //并不是每一行访问行为都有searchKeyword和clickCategoryId两个字段的
                            //其实，只有搜索行为是有searchKeyword字段的
                            //只有点击品类的行为是有clickCaregoryId字段的
                            //所以，任何一行行为数据，都不可能两个字段都有，所以数据是可能出现null值的
                            //所以是否将搜索词点击品类id拼接到字符串中去
                            //首先要满足不能是null值
                            //其次，之前的字符串中还没有搜索词或者点击品类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 + ",");
                                }
                            }                       }
                        //StringUtils引入的包是import com.erik.sparkproject.util.trimComma;
                        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>
                        //然后在直接将返回的Tuple的key设置成sessionid
                        //最后的数据格式，还是<sessionid,fullAggrInfo>
                        //聚合数据，用什么样的格式进行拼接？
                        //我们这里统一定义，使用key=value|key=vale
                        //在Constants.java中定义spark作业相关的常量
                        //String FIELD_SESSION_ID = "sessionid";
                        //String FIELD_SEARCH_KEYWORDS = "searchKeywords";
                        //String FIELD_CLICK_CATEGORY_IDS = "clickCategoryIds";
                        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);
                    }
                });
        //将session粒度聚合数据，与用户信息进行join
        JavaPairRDD<Long, Tuple2<String, Row>> userid2FullInfoRDD = 
                userid2PartAggrInfoRDD.join(userid2InfoRDD);
        //对join起来的数据进行拼接，并且返回<sessionid,fullAggrInfo>格式的数据
        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);
                        //在Constants.java中添加以下常量
                        //String FIELD_AGE = "age";
                        //String FIELD_PROFESSIONAL = "professional";
                        //String FIELD_CITY = "city";
                        //String FIELD_SEX = "sex";
                        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数据
     * @param sessionid2AggrInfoRDD
     * @return
     */
    private static JavaPairRDD<String, String> filterSession(
            JavaPairRDD<String, String> sessionid2AggrInfoRDD, 
            final JSONObject taskParam) {
        //为了使用后面的ValieUtils,所以，首先将所有的筛选参数拼接成一个连接串
        String startAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE);
        String endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE);
        String professionals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS);
        String cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITIES);
        String sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX);
        String keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS);
        String categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS);
        String _parameter = (startAge != null ? Constants.PARAM_START_AGE + "=" + startAge + "|" : "")
                + (endAge != null ? Constants.PARAM_END_AGE + "=" + endAge + "|" : "")
                + (professionals != null ? Constants.PARAM_PROFESSIONALS + "=" + professionals + "|" : "")
                + (cities != null ? Constants.PARAM_CITIES + "=" + cities + "|" : "")
                + (sex != null ? Constants.PARAM_SEX + "=" + sex + "|" : "")
                + (keywords != null ? Constants.PARAM_KEYWORDS + "=" + keywords + "|" : "")
                + (categoryIds != null ? Constants.PARAM_CATEGORY_IDS + "=" + categoryIds : "");
        if (_parameter.endsWith("\\|")) {
            _parameter = _parameter.substring(0, _parameter.length() - 1);
        }
        final String parameter = _parameter;
        //根据筛选参数进行过滤
        JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD = sessionid2AggrInfoRDD.filter(
                new Function<Tuple2<String, String>, Boolean>() {
                    private static final long serialVersionUID = 1L;
                    public Boolean call(Tuple2<String, String> tuple) throws Exception {
                        //首先，从tuple中，获取聚合数据
                        String aggrInfo = tuple._2;
                        //接着，依次按照筛选条件进行过滤
                        //按照年龄范围进行过滤（startAge、endAge）
                        //先在Constants.java中添加常量
                        //String PARAM_START_AGE = "startAge";
                        //String PARAM_END_AGE = "endage";
                        //String PARAM_PROFESSIONALS = "professionals";
                        //String PARAM_CITIES = "cities";
                        //String PARAM_SEX = "sex";
                        //String PARAM_KEYWORDS = "keywords";
                        //String PARAM_CATEGORY_IDS = "categoryIds";
                        if(!ValidUtils.between(aggrInfo, Constants.FIELD_AGE, 
                                parameter, Constants.PARAM_START_AGE, Constants.PARAM_END_AGE)) {
                            return false;
                        }
                        //按照职业范围进行过滤（professionals）
                        if(!ValidUtils.in(aggrInfo, Constants.FIELD_PROFESSIONAL, 
                                parameter, Constants.PARAM_PROFESSIONALS)) {
                            return false;
                        }
                        //按照城市范围进行过滤（cities）
                        if(!ValidUtils.in(aggrInfo, Constants.FIELD_CITY, 
                                parameter, Constants.PARAM_CATEGORY_IDS)) {
                            return false;
                        }
                        //按照性别过滤
                        if(!ValidUtils.equal(aggrInfo, Constants.FIELD_SEX, 
                                parameter, Constants.PARAM_SEX)) {
                            return false;
                        }
                        //按照搜索词过滤
                        if(!ValidUtils.in(aggrInfo, Constants.FIELD_SEARCH_KEYWORDS, 
                                parameter, Constants.PARAM_KEYWORDS)) {
                            return false;                           
                        }
                        //按照点击品类id进行搜索
                        if(!ValidUtils.in(aggrInfo, Constants.FIELD_CLICK_CATEGORY_IDS, 
                                parameter, Constants.PARAM_CATEGORY_IDS)) {
                            return false;
                        }                   
                        return true;
                    }       
        });     
        return null;
    }   
}

session聚合统计之自定义Accumulator

之前已经得到通过条件过滤后的session，现在需要统计出来访问时长在0s~3s的session的数量，占总session数量的比例；4s~6s的session的数量，占总session数量的比例；……
并且统计出访问步长在1~3的session的数量，占总session数量的比例；4~6的session的数量，占总session数量的比例；……

这是就需要Accumulator，传统的Accumulator实现方式为：
Accumulator 1s_3s = sc.accumulator(0L);
十几个Accumulator

可以对过滤以后的session，调用foreach也可以，遍历所有session；计算每个session的访问时长和访问步长；
访问时长：把session的最后一个action的时间，减去第一个action的时间。
访问步长：session的action数量。
计算出访问时长和访问步长以后，根据对应的区间，找到对应的Accumulator，1s_3s.add(1L)，
同时每遍历一个session，就可以给总session数量对应的Accumulator加1，最后用各个区间的session数量，除以总session数量，就可以计算出各个区间的占比了。

这种传统的实现方式，有什么不好呢？？？

最大的不好，就是Accumulator太多了，不便于维护。
首先第一，很有可能，在写后面的累加代码的时候，比如找到了一个4s~6s的区间的session，但是却代码里面不小心，累加到7s~9s里面去了；
第二，当后期，项目如果要出现一些逻辑上的变更，比如说，session数量的计算逻辑，要改变，就得更改所有Accumulator对应的代码；或者说，又要增加几个范围，那么又要增加多个Accumulator，并且修改对应的累加代码；维护成本，相当之高（甚至可能，修改一个小功能，或者增加一个小功能，耗费的时间，比做一个新项目还要多；甚至于，还修改出了bug，那就耗费更多的时间）。

所以，我们这里的设计，不打算采用传统的方式，用十几个，甚至二十个Accumulator，因为维护成本太高。
这里的实现思路是，我们自己自定义一个Accumulator，实现较为复杂的计算逻辑，一个Accumulator维护了所有范围区间的数量的统计逻辑。
低耦合，如果说，session数量计算逻辑要改变，那么不用变更session遍历的相关的代码；只要维护一个Accumulator里面的代码即可；如果计算逻辑后期变更，或者加了几个范围，那么也很方便，不用多加好几个Accumulator，去修改大量的代码；只要维护一个Accumulator里面的代码即可；维护成本，大大降低

自定义Accumulator，也是Spark Core中，属于比较高端的一个技术使用自定义Accumulator，大家就可以任意的实现自己的复杂分布式计算的逻辑。如果说，你的task，分布式，进行复杂计算逻辑，那么是很难实现的（借助于redis，维护中间状态，借助于zookeeper去实现分布式锁）。但是，使用自定义Accumulator，可以更方便进行中间状态的维护，而且不用担心并发和锁的问题。

package com.erik.sparkproject.spark;
import org.apache.spark.AccumulatorParam;
import com.erik.sparkproject.constant.Constants;
import com.erik.sparkproject.util.StringUtils;
/**
 * session聚合统计Accumulator
 * 使用自己定义的一些数据格式，比如String，甚至说，我们可以自己定义model，自己定义的类（必须可序列化）
 * 然后可以基于这种特殊的数据格式实现复杂的分布式计算逻辑
 * 各个task分布式在运行，可以根据自己的需求，task给Accumulator传入不同的值
 * 根据不同的值，去做复杂的逻辑
 * 
 * Spark Core里很使用的高端技术
 * @author Erik
 *
 */
public class SesssionAggrStatAccumulator implements AccumulatorParam<String> {
    private static final long serialVersionUID = -2113961376143864034L;
    //zero方法，主要用于数据初始化
    //这里就返回一个值，就是初始化中，所有范围区间的数值都是0
    //各个范围区间的统计数量的拼接，还是采用key=value|key=value的连接串格式
    //在Constants.java中添加
    //String SESSION_COUNT = "session_count";
    //String TIME_PERIOD_1s_3s = "1s_3s";
    //String TIME_PERIOD_4s_6s = "4s_6s";
    //String TIME_PERIOD_7s_9s = "7s_9s";
    //String TIME_PERIOD_10s_30s = "10s_30s";
    //String TIME_PERIOD_30s_60s = "30s_60s";
    //String TIME_PERIOD_1m_3m = "1m_3m";
    //String TIME_PERIOD_3m_10m = "3m_10m";
    //String TIME_PERIOD_10m_30m = "10m_30m";
    //String TIME_PERIOD_30m = "30m";
    //String STEP_PERIOD_1_3 = "1_3";
    //String STEP_PERIOD_4_6 = "4_6";
    //String STEP_PERIOD_7_9 = "7_9";
    //String STEP_PERIOD_10_30 = "10_30";
    //String STEP_PERIOD_30_60 = "30_60"; 
    //String STEP_PERIOD_60 = "60";
    public String zero(String v) {
        return Constants.SESSION_COUNT + "=0|"
                + Constants.TIME_PERIOD_1s_3s + "=0|"
                + Constants.TIME_PERIOD_4s_6s + "=0|"
                + Constants.TIME_PERIOD_7s_9s + "=0|"
                + Constants.TIME_PERIOD_10s_30s + "=0|"
                + Constants.TIME_PERIOD_30s_60s + "=0|"
                + Constants.TIME_PERIOD_1m_3m + "=0|"
                + Constants.TIME_PERIOD_3m_10m + "=0|"
                + Constants.TIME_PERIOD_10m_30m + "=0|"
                + Constants.TIME_PERIOD_30m + "=0|"
                + Constants.STEP_PERIOD_1_3 + "=0|"
                + Constants.STEP_PERIOD_4_6 + "=0|"
                + Constants.STEP_PERIOD_7_9 + "=0|"
                + Constants.STEP_PERIOD_10_30 + "=0|"
                + Constants.STEP_PERIOD_30_60 + "=0|"
                + Constants.STEP_PERIOD_60 + "=0";
    }
    //addInPlace和addAccumulator可以理解为是一样的
    //这两个方法，主要实现，v1可能就是我们初始化的那个连接串
    //v2就是在遍历session的时候，判断出某个session对应的区间，然后用Constants.TIME_PERIOD_1S_3S
    //所以，我们要做的事情就是在v1中，找到对应的v2对应的value，累加1，然后再更新到连接串里面去
    public String addInPlace(String v1, String v2) {
        return null;
    }
    public String addAccumulator(String v1, String v2) {
        return null;
    }
    /**
     * session统计计算逻辑
     * @param v1连接串
     * @param v2 范围区间
     * @return更新以后的连接串
     */
    private String add(String v1, String v2) {
        //校验：v1为空的时候，直接返回v2
        if(StringUtils.isEmpty(v1)) {
            return v2;
        }
        //使用StringUtils工具类，从v1中提取v2对应的值，并累加1
        String oldValue = StringUtils.getFieldFromConcatString(v1, "\\|", v2);
        if(oldValue != null) {
            //将范围区间原有的值累加1
            int newValue = Integer.valueOf(oldValue) + 1;
            //使用StringUtils工具类，将v1中，v2对应的值设置成心的累加胡的值
            return StringUtils.setFieldInConcatString(v1, "\\|", v2, String.valueOf(newValue));
        }
        return v1;
    }
}

session聚合统计之重构实现思路与重构session聚合

接下来需要进行session聚合统计，统计出访问时长和访问步长，各个区间的session数量占总session数量的比例。

如果不进行重构，直接来实现，思路如下：
1. sessionRDD，映射成的格式；
2. 按session聚合，计算出每个session的访问时长和访问步长，生成一个新的RDD；
3. 遍历新生成的RDD，将每个session的访问时长和访问步长，去更新自定义Accumulator中对应的值；
4. 使用自定义Accumulator中的统计值，去计算各个区间的比例；
5. 将最后计算出来的结果，写入MySQL对应的表中。

普通实现思路的问题：
1. 为什么还要用actionRDD去映射？其实之前在session聚合的时候已经做过映射了，多次一举;
2. 是不是一定要为了session的聚合这个功能单独去遍历一遍session？其实没必要，已经有session数据，之前过滤session的时候，其实相当于是在遍历session了，那么这里就没必要再过滤一遍了。

重构实现思路
不要去生成任何新的RDD（处理上亿的数据）；
不要去单独遍历一遍session（处理上千万的数据）；
可以在进行session聚合的时候就直接计算出来每个session的访问时长和访问步长；
在进行过滤的时候，本来就要遍历所有的聚合session信息，此时就可以在某个session通过筛选条件后 将访问时长和访问步长累加到自定义的Acccumulator上面；
就是两种截然不同的思考方式和实现方式，在面对上亿，上千万数据的时候，甚至可以节省时间长达半小时 或者数个小时。
开发Spark大型复杂项目的一些经验准则
尽量少生成RDD；
尽量少对RDD进行算子操作，如果有可能，尽量在一个算子里实现多个需要的功能；
尽量少对进行RDD进行shuffle算子操作，比如groupByKey,reduceByKey,sortByKey等shuffle操作，会导致大量的磁盘读写，严重降低性能。有shuffle的算子和没有shuffle的算子，性能甚至会有长达几十分钟甚至数个小时的差别，有shuffle的算子，很容易导致数据倾斜，一旦数据倾斜，简直就是性能杀手（后续会有完整解决方案）
无论做什么功能，性能第一。
在传统的J2EE或者.NET或者PHP，软件/系统/网站开发中，架构和可维护性，可扩展性的重要程度远远高于性能，大量的分布式的架构，设计模式，代码的划分，类的划分（高并发网站除外）在大数据项目中，比如MapReduce，Hive，Spark,Storm中，性能的重要程度远远大于一次额代码的规范和设计模式，代码的划分，类的划分；大数据最重要的是性能。

主要就是因为大数据以及大数据项目的特点决定了大数据的程度和项目的速度都比较慢，如果不优先考虑性能的话，会导致一个大数据处理程序运行时间长达数小时，甚至数十个小时，所以，推荐大数据项目，在开发和代码的架构中，优先考虑性能，其次考虑功能代码的划分、解耦合。

我们如果采用第一种实现方案，那么其实就是代码划分（解耦合、可维护）优先，设计优先如果采用第二种方案，那么其实就是性能优先。

package com.erik.sparkproject.spark;
import java.text.ParseException;
import java.util.Date;
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.Function;
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;
/**
 * 用户访问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 Erik
 *
 */
public class UserVisitSessionAnalyzeSpark {
    public static void main(String[] args) {
        args = new String[]{"2"}; 
        //构建spark上下文
        //首先在Constants.java中设置spark作业相关的常量
        //String SPARK_APP_NAME = "UserVisitSessionAnalyzeSpark";
        //保存Constants.java配置
        SparkConf conf = new SparkConf()
                .setAppName(Constants.SPARK_APP_NAME)
                .setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = getSQLContext(sc.sc());
        //生成模拟测试数据
        mockData(sc, sqlContext);
        //创建需要使用的DAO组件
        ITaskDAO taskDAO = DAOFactory.getTaskDAO();
        //那么就首先得查询出来指定的任务，并获取任务的查询参数
        long taskid = ParamUtils.getTaskIdFromArgs(args);
        Task task = taskDAO.findById(taskid);
        JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());
        //如果要进行session粒度的数据聚合，
        //首先要从user_visit_action表中，查询出来指定日期范围内的数据
        JavaRDD<Row> actionRDD = getActionRDDByDateRange(sqlContext, taskParam);
        //聚合
        //首先，可以将行为数据按照session_id进行groupByKey分组
        //此时的数据粒度就是session粒度了，然后可以将session粒度的数据与用户信息数据惊醒join
        //然后就可以获取到session粒度的数据，同时数据里面还包含了session对应的user信息
        //到这里为止，获取的数据是<sessionid,(sessionid,searchKeywords,
        //clickCategoryIds,age,professional,city,sex)>
        JavaPairRDD<String, String> sessionid2AggrInfoRDD = 
                aggregateBySession(sqlContext, actionRDD);
        //接着，就要针对session粒度的聚合数据，按照使用者指定的筛选参数进行数据过滤
        //相当于我们自己编写的算子，是要访问外面的任务参数对象的
        //匿名内部类（算子函数），访问外部对象，是要给外部对象使用final修饰的
        JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD = 
                filterSession(sessionid2AggrInfoRDD, taskParam);
        /**
         * session聚合统计（统计出访问时长和访问步长，各个区间的session数量占总session数量的比例）
         */
        //关闭spark上下文
        sc.close();
    }
    /**
     * 获取SQLContext
     * 如果在本地测试环境的话，那么久生成SQLC哦那text对象
     *如果在生产环境运行的话，那么就生成HiveContext对象
     * @param sc SparkContext
     * @return SQLContext
     */
    private static SQLContext getSQLContext(SparkContext sc) {
        //在my.properties中配置
        //spark.local=true（打包之前改为flase）
        //在ConfigurationManager.java中添加
        //public static Boolean getBoolean(String key) {
        //  String value = getProperty(key);
        //  try {
        //      return Boolean.valueOf(value);
        //  } catch (Exception e) {
        //      e.printStackTrace();
        //  }
        //  return false;   
        //}
        //在Contants.java中添加
        //String SPARK_LOCAL = "spark.local";
        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) {
        //先在Constants.java中添加任务相关的常量
        //String PARAM_START_DATE = "startDate";
        //String PARAM_END_DATE = "endDate";
        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) {
        //现在actionRDD中的元素是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;
                    public Tuple2<String, Row> call(Row row) throws Exception {
                        //按照MockData.java中字段顺序获取
                        //此时需要拿到session_id，序号是2
                        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;
                    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;
                        //session的起始和结束时间
                        Date startTime = null;
                        Date endTime = null;
                        //session的访问步长
                        int stepLength = 0;
                        //遍历session所有的访问行为
                        while(iterator.hasNext()) {
                            //提取每个 访问行为的搜索词字段和点击品类字段
                            Row row = iterator.next();
                            if(userid == null) {
                                userid = row.getLong(1);
                            }
                            String searchKeyword = row.getString(5);
                            Long clickCategoryId = row.getLong(6);
                            //实际上这里要对数据说明一下
                            //并不是每一行访问行为都有searchKeyword和clickCategoryId两个字段的
                            //其实，只有搜索行为是有searchKeyword字段的
                            //只有点击品类的行为是有clickCaregoryId字段的
                            //所以，任何一行行为数据，都不可能两个字段都有，所以数据是可能出现null值的
                            //所以是否将搜索词点击品类id拼接到字符串中去
                            //首先要满足不能是null值
                            //其次，之前的字符串中还没有搜索词或者点击品类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 + ",");
                                }
                            }
                            //计算session开始和结束时间
                            Date actionTime = DateUtils.parseTime(row.getString(4));
                            if(startTime == null) {
                                startTime = actionTime;
                            }
                            if(endTime == null) {
                                endTime = actionTime;
                            }
                            if(actionTime.before(startTime)) {
                                startTime = actionTime;
                            }
                            if(actionTime.after(endTime)) {
                                endTime = actionTime;
                            }
                            //计算session访问步长
                            stepLength ++;
                        }
                        //计算session开始和结束时间
                        //现在DateUtils.java中添加方法
                        //public static Date parseTime(String time) {
                        //  try {
                        //      return TIME_FORMAT.parse(time);
                        //  } catch (ParseException e) {
                        //      e.printStackTrace();
                        //  }
                        //  return null;
                        //}
                        //StringUtils引入的包是import com.erik.sparkproject.util.trimComma;
                        String searchKeywords = StringUtils.trimComma(searchKeywordsBuffer.toString());
                        String clickCategoryIds = StringUtils.trimComma(clickCategoryIdsBuffer.toString());
                        //计算session访问时长（秒）
                        long visitLength = (endTime.getTime() - startTime.getTime()) / 1000;
                        //返回的数据即是<sessionid, partAggrInfo>
                        //但是，这一步聚合后，其实还需要将每一行数据，根对应的用户信息进行聚合
                        //问题来了，如果是跟用户信息进行聚合的话，那么key就不应该是sessionid，而应该是userid
                        //才能够跟<userid, Row>格式的用户信息进行聚合
                        //如果我们这里直接返回<sessionid, partAggrInfo>,还得再做一次mapToPair算子
                        //将RDD映射成<userid,partAggrInfo>的格式，那么就多此一举
                        //所以，我们这里其实可以直接返回数据格式就是<userid,partAggrInfo>
                        //然后在直接将返回的Tuple的key设置成sessionid
                        //最后的数据格式，还是<sessionid,fullAggrInfo>
                        //聚合数据，用什么样的格式进行拼接？
                        //我们这里统一定义，使用key=value|key=vale
                        //在Constants.java中定义spark作业相关的常量
                        //String FIELD_SESSION_ID = "sessionid";
                        //String FIELD_SEARCH_KEYWORDS = "searchKeywords";
                        //String FIELD_CLICK_CATEGORY_IDS = "clickCategoryIds";
                        //String FIELD_VISIT_LENGTH = "visitLength";
                        //String FIELD_STEP_LENGTH = "stepLength";
                        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<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);
                    }               
                });
        //将session粒度聚合数据，与用户信息进行join
        JavaPairRDD<Long, Tuple2<String, Row>> userid2FullInfoRDD = 
                userid2PartAggrInfoRDD.join(userid2InfoRDD);
        //对join起来的数据进行拼接，并且返回<sessionid,fullAggrInfo>格式的数据
        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);
                        //在Constants.java中添加以下常量
                        //String FIELD_AGE = "age";
                        //String FIELD_PROFESSIONAL = "professional";
                        //String FIELD_CITY = "city";
                        //String FIELD_SEX = "sex";
                        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数据
     * @param sessionid2AggrInfoRDD
     * @return
     */
    private static JavaPairRDD<String, String> filterSession(
            JavaPairRDD<String, String> sessionid2AggrInfoRDD, 
            final JSONObject taskParam) {
        //为了使用后面的ValieUtils,所以，首先将所有的筛选参数拼接成一个连接串
        String startAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE);
        String endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE);
        String professionals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS);
        String cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITIES);
        String sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX);
        String keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS);
        String categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS);
        String _parameter = (startAge != null ? Constants.PARAM_START_AGE + "=" + startAge + "|" : "")
                + (endAge != null ? Constants.PARAM_END_AGE + "=" + endAge + "|" : "")
                + (professionals != null ? Constants.PARAM_PROFESSIONALS + "=" + professionals + "|" : "")
                + (cities != null ? Constants.PARAM_CITIES + "=" + cities + "|" : "")
                + (sex != null ? Constants.PARAM_SEX + "=" + sex + "|" : "")
                + (keywords != null ? Constants.PARAM_KEYWORDS + "=" + keywords + "|" : "")
                + (categoryIds != null ? Constants.PARAM_CATEGORY_IDS + "=" + categoryIds : "");
        if (_parameter.endsWith("\\|")) {
            _parameter = _parameter.substring(0, _parameter.length() - 1);
        }
        final String parameter = _parameter;
        //根据筛选参数进行过滤
        JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD = sessionid2AggrInfoRDD.filter(
                new Function<Tuple2<String, String>, Boolean>() {
                    private static final long serialVersionUID = 1L;
                    public Boolean call(Tuple2<String, String> tuple) throws Exception {
                        //首先，从tuple中，获取聚合数据
                        String aggrInfo = tuple._2;
                        //接着，依次按照筛选条件进行过滤
                        //按照年龄范围进行过滤（startAge、endAge）
                        //先在Constants.java中添加常量
                        //String PARAM_START_AGE = "startAge";
                        //String PARAM_END_AGE = "endage";
                        //String PARAM_PROFESSIONALS = "professionals";
                        //String PARAM_CITIES = "cities";
                        //String PARAM_SEX = "sex";
                        //String PARAM_KEYWORDS = "keywords";
                        //String PARAM_CATEGORY_IDS = "categoryIds";
                        if(!ValidUtils.between(aggrInfo, Constants.FIELD_AGE, 
                                parameter, Constants.PARAM_START_AGE, Constants.PARAM_END_AGE)) {
                            return false;
                        }
                        //按照职业范围进行过滤（professionals）
                        if(!ValidUtils.in(aggrInfo, Constants.FIELD_PROFESSIONAL, 
                                parameter, Constants.PARAM_PROFESSIONALS)) {
                            return false;
                        }
                        //按照城市范围进行过滤（cities）
                        if(!ValidUtils.in(aggrInfo, Constants.FIELD_CITY, 
                                parameter, Constants.PARAM_CATEGORY_IDS)) {
                            return false;
                        }
                        //按照性别过滤
                        if(!ValidUtils.equal(aggrInfo, Constants.FIELD_SEX, 
                                parameter, Constants.PARAM_SEX)) {
                            return false;
                        }
                        //按照搜索词过滤
                        if(!ValidUtils.in(aggrInfo, Constants.FIELD_SEARCH_KEYWORDS, 
                                parameter, Constants.PARAM_KEYWORDS)) {
                            return false;
                        }
                        //按照点击品类id进行搜索
                        if(!ValidUtils.in(aggrInfo, Constants.FIELD_CLICK_CATEGORY_IDS, 
                                parameter, Constants.PARAM_CATEGORY_IDS)) {
                            return false;
                        }                   
                        return true;
                    }       
        }); 
        return null;
    }
}