按时间比例随机抽取session

每一次执行用户访问session分析模块，要抽取出100个session

session随机抽取：按每天的每个小时的session数量，占当天session总数的比例，乘以每天要抽取的session数量，计算出每个小时要抽取的session数量；然后呢，在每天每小时的session中，随机抽取出之前计算出来的数量的session。

举例：10000个session，抽取100个session；0点~1点之间，有2000个session，占总session的比例就是0.2；按照比例，0点~1点需要抽取出来的session数量是100 * 0.2 = 20个；在0点~点的2000个session中，随机抽取出来20个session。

1.1、我们之前有什么数据：session粒度的聚合数据（计算出来session的start_timesession聚合数据进行映射，将每个session发生的yyyy-MM-dd_HH（start_time）作为key，value就是聚合的数据

    //
        JavaPairRDD session2TimeRDD = filteredSessionid2AggrInfoRDD.mapToPair(new PairFunction() {

            @Override
            public Tuple2 call(Tuple2 t)
                    throws Exception {
                String aggrInfo = t._2;
                //获取session的开始时间
                String startTime =  StringUtils.getFieldFromConcatString(
                        aggrInfo, "\\|", Constants.FIELD_START_TIME);
                String dateHour = DateUtils.getDateHour(startTime);

                return new Tuple2(dateHour, aggrInfo);
            }
        });

1.2对上述数据，使用countByKey算子，就可以获取到每天每小时的session数量

        //统计出每小时的session量
        Map countMap = session2TimeRDD.countByKey();
        // 第二步，使用按时间比例随机抽取算法，计算出每天每小的session数量
        // 将格式的map，转换成的格式
        Map dateHourCountMap = 
                new HashMap();

        for(Map.Entry countEntry : countMap.entrySet()) {
            String dateHour = countEntry.getKey();
            String date = dateHour.split("_")[0];
            String hour = dateHour.split("_")[1];  

            long count = Long.valueOf(String.valueOf(countEntry.getValue()));  

            Map hourCountMap = dateHourCountMap.get(date);
            if(hourCountMap == null) {
                hourCountMap = new HashMap();
                dateHourCountMap.put(date, hourCountMap);
            }

            hourCountMap.put(hour, count);
        }
        System.out.println(dateHourCountMap);

1.3、（按时间比例随机抽取算法）每天每小时有多少session， 1.3.1、根据这个数量计算出每天每小时的session占比， 1.3.2、按照占比，需要抽取多少session，可以计算出每个小时内，从0~session数量之间的范围中，获取指定抽取数量个随机数，作为随机抽取的索引

            // 开始实现我们的按时间比例随机抽取算法

            // 总共要抽取100个session，先按照天数，进行平分
            int extractNumberPerDay = 100 / dateHourCountMap.size();

            //   
            Map dateHourExtractMap = 
                    new HashMap();

            Random random = new Random();
            for(Map.Entry dateHourCountEntry : dateHourCountMap.entrySet()) {
                String date = dateHourCountEntry.getKey();
                Map hourCountMap = dateHourCountEntry.getValue();

                // 计算出这一天的session总数
                long sessionCount = 0L;
                for(long hourCount : hourCountMap.values()) {
                    sessionCount += hourCount;
                }

                Map hourExtractMap = dateHourExtractMap.get(date);
                if(hourExtractMap == null) {
                    hourExtractMap = new HashMap();
                    dateHourExtractMap.put(date, hourExtractMap);
                }

                // 遍历每个小时
                for(Map.Entry hourCountEntry : hourCountMap.entrySet()) {
                    String hour = hourCountEntry.getKey();
                    long count = hourCountEntry.getValue();//每小时的session量

                    // 计算每个小时的session数量，占据当天总session数量的比例，直接乘以每天要抽取的数量
                    // 就可以计算出，当前小时需要抽取的session数量
                    int hourExtractNumber = (int)(((double)count / (double)sessionCount) 
                            * extractNumberPerDay);
                    if(hourExtractNumber > count) {
                        hourExtractNumber = (int) count;
                    }

                    // 先获取当前小时的存放随机数的list
                    List extractIndexList = hourExtractMap.get(hour);
                    if(extractIndexList == null) {
                        extractIndexList = new ArrayList();
                        hourExtractMap.put(hour, extractIndexList);
                    }

                    // 生成上面计算出来的数量的随机数
                    for(int i = 0; i