Spark实践 -- 夜出顾客服务分析

1 业务需求

最近做的24小时书店大数据平台中的一个需求：获取一段时间内只在晚上进店，而白天没有进店的顾客。
输入是指定的开始日期、结束日期、夜出开始时间(nightTimeS)、夜出结束时间(nightTimeE)。通过userName可以区分一个顾客。

2 业务实现

2.1 第一版只统计了晚上出现的顾客

下面代码是最开始的实现，有些问题需要改进：

只将晚上出现过的客户统计起来，而没考虑该顾客可能白天也出现过，基本没有实现业务需求；
另外对传入的夜出时间范围的判定也不够严谨。另外对传入的夜出时间范围的判定也不够严谨。

// 得到<userName_date, count>
JavaPairRDD<String, Integer> pairRdd = df.toJavaRDD().mapToPair(new PairFunction<Row, String, Integer>() {
    @Override
    public Tuple2<String, Integer> call(Row row) throws Exception {
        String userName = row.getString(0);
        String[] strAry = Strsplit.splitByWholeSeparator(row.getString(1), " ", -1, true);
        if(strAry[1].compareTo(nightTimeS)>0){
            return new Tuple2<String, Integer>(String.format("%s,%s",userName, strAry[0]), 1);
        } else if(strAry[1].compareTo(nightTimeE)<0){
            Date preDate = DateUtil.parse(strAry[0], "yyyy-MM-dd");
            String preDate_s = DateUtil.format(DateUtil.addDays(preDate, -1), "yyyy-MM-dd");
            return new Tuple2<String, Integer>(String.format("%s,%s",userName, preDate_s), 1);
        } else {
            return new Tuple2<String, Integer>(String.format("%s,%s",userName, strAry[0]), 0);
        }
    }
});
// 将userName_date相同的相加
JavaPairRDD<String, Integer> pairRdd2 = pairRdd.reduceByKey(new Function2<Integer, Integer, Integer>() {
    @Override
    public Integer call(Integer v1, Integer v2) throws Exception {
        return v1 + v2;
    }
});
// 过滤出现次数小于1的
JavaPairRDD<String, Integer> pairRdd3 = pairRdd2.filter(new Function<Tuple2<String,Integer>, Boolean>() {
    @Override
    public Boolean call(Tuple2<String, Integer> v1) throws Exception {
        return v1._2 > 0;
    }
});
// 结果处理
JavaRDD<String> result = pairRdd3.map(new Function<Tuple2<String,Integer>, String>() {
    @Override
    public String call(Tuple2<String, Integer> tuple) throws Exception {
        String[] strAry = Strsplit.splitByWholeSeparator(tuple._1, ",", -1, true);
        String dayEndDate = DateUtil.format(DateUtil.addDays(DateUtil.parse(strAry[1], "yyyy-MM-dd"),1), "yyyy-MM-dd");
        return String.format("%s,%s,%s,%s", strAry[0],strAry[1],dayEndDate,tuple._2);
    }
});

2.2 第二版对白天进店了的顾客形成列表然后用于后续过滤

该版实现做的工作：

增加了对夜出范围更完整的逻辑判断。
使用一个List来保存白天出现过的顾客，然后再通过这个List来把结果中在白天出现过且晚上有出现的顾客过滤掉。

经过线上测试，该版本性能极低。因为List中会存储所有白天出现的顾客导致过滤过程缓慢。由于分布式的原因，List还无法使用移除里面的null值。

// 获取白天出现过的顾客列表
final JavaRDD<String> rdd = df.toJavaRDD().map(new Function<Row, String>() {
    @Override
    public String call(Row row) throws Exception {
        String[] strAry = Strsplit.splitByWholeSeparator(row.getString(1), " ", -1, true);
        // 下面第一重if是判断夜晚出现的范围， 第二重if是判断顾客在不在范围内
        // nightTimeS  < nightTimeE
        if(nightTimeS.compareTo(nightTimeE) < 0){
            if(strAry[1].compareTo(nightTimeS) > 0 && strAry[1].compareTo(nightTimeE) < 0){
                return "0";
            }else{
                return row.getString(0);
            }
        }else{ // nightTimeS  > nightTimeE     (包括三种情况：1，22：00 - 00：00  2，00：00 - 02：00  3. 22：00 - 02：00  其实也可以归为一种情况，前两种都是第三种的特例）
            if(strAry[1].compareTo(nightTimeS)> 0 || strAry[1].compareTo(nightTimeE) < 0){
                return "0";
            } else {
                return row.getString(0);
            }
        }
    }
});
final List<String> userNameList = rdd.toArray();

JavaPairRDD<String, Integer> pairRddbefore = df.toJavaRDD().mapToPair(new PairFunction<Row, String, Integer>() {
    @Override
    public Tuple2<String, Integer> call(Row row) throws Exception {
        //row ： user1  2018-10-11 21:11:11  然后将时间切分成： strAry[0] 2018-10-11    strAry[1] 21:11:11
        String userName = row.getString(0);
        String[] strAry = Strsplit.splitByWholeSeparator(row.getString(1), " ", -1, true);

        if(nightTimeS.compareTo(nightTimeE) < 0){
            if(strAry[1].compareTo(nightTimeS) > 0 && strAry[1].compareTo(nightTimeE) < 0){
                // 区分在 00:00:00前 与 00:00:00后的
                //  00:00:00前
                if(strAry[1].compareTo("24:00:00") < 0 && strAry[1].compareTo("12:00:00") > 0){
                    return new Tuple2<String, Integer>(String.format("%s,%s",userName,strAry[0]), 1);
                }else{  //  00:00:00 后
                    Date preDate = DateUtil.parse(strAry[0], "yyyy-MM-dd");
                    String preDate_s = DateUtil.format(DateUtil.addDays(preDate, -1), "yyyy-MM-dd");
                    return new Tuple2<String, Integer>(String.format("%s,%s",userName, preDate_s), 1);
                }
            }else{
                return new Tuple2<String, Integer>(String.format("%s,%s", userName, strAry[0]), 0);
            }
        }else{ // nightTimeS  > nightTimeE
            if(strAry[1].compareTo(nightTimeS) > 0){
                return new Tuple2<String, Integer>(String.format("%s,%s", userName, strAry[0]), 1);
            } else if(strAry[1].compareTo(nightTimeE) < 0){
                Date preDate = DateUtil.parse(strAry[0], "yyyy-MM-dd");
                String preDate_s = DateUtil.format(DateUtil.addDays(preDate, -1), "yyyy-MM-dd");
                return new Tuple2<String, Integer>(String.format("%s,%s", userName, preDate_s), 1);
            } else {
                return new Tuple2<String, Integer>(String.format("%s,%s", userName, strAry[0]), 0);
            }
        }
    }
});
// 执行过滤，pairRddbefore包含全部夜晚出现过的顾客，
JavaPairRDD<String, Integer> pairRdd = pairRddbefore.filter(new Function<Tuple2<String, Integer>, Boolean>(){
    @Override
    public Boolean call(Tuple2<String, Integer> tuple) throws Exception {
        String userName = tuple._1.split(",")[0];
        return !userNameList.contains(userName);
    }
});

2.3 第三版通过求子集过滤掉白天出现过的所有顾客

由于第二版的实现中的List太过冗余，所以第三版的实现在这问题上做了改进。
通过subtractByKey算子过滤掉白天出现过的所有顾客。然后进行后续的处理中，计算的就都是在夜出范围内且白天没出现过的顾客了。

// 获取白天出现过的顾客列表
JavaPairRDD<String, String> rdd = df.toJavaRDD().mapToPair(new PairFunction<Row, String, String>() {
    @Override
    public Tuple2<String, String> call(Row row) throws Exception {
        String[] strAry = Strsplit.splitByWholeSeparator(row.getString(1), " ", -1, true);
        // 下面第一重if是判断 昼伏夜出的范围， 第二重if是判断车辆在不在范围内
        // nightTimeS  < nightTimeE
        if(nightTimeS.compareTo(nightTimeE) < 0){
            if(strAry[1].compareTo(nightTimeS) > 0 && strAry[1].compareTo(nightTimeE) < 0){
                return new Tuple2<>("0", "0");
            }else{
                return new Tuple2<>(row.getString(0), row.getString(1));
            }
        }else{ // nightTimeS  > nightTimeE     (包括三种情况：1，22：00 - 00：00  2，00：00 - 02：00  3. 22：00 - 02：00  其实也可以归为一种情况，前两种都是第三种的特例）
            if(strAry[1].compareTo(nightTimeS)> 0 || strAry[1].compareTo(nightTimeE) < 0){
                return new Tuple2<>("0", "0");
            } else {
                return new Tuple2<>(row.getString(0), row.getString(1));
            }
        }
    }
});

// 获取全部顾客列表
JavaPairRDD<String, String> rdd2 = df.toJavaRDD().mapToPair(new PairFunction<Row, String, String>() {
    @Override
    public Tuple2<String, String> call(Row row) throws Exception {
        return new Tuple2<>(row.getString(0), row.getString(1));
    }
});

// 做差集，获得只在所求的夜出时间段内的顾客名及时间  <userName, capDate>
JavaPairRDD<String, String> rdd3 = rdd2.subtractByKey(rdd);

JavaPairRDD<String, Integer> pairRdd = rdd3.mapToPair(new PairFunction<Tuple2<String, String>, String, Integer>() {
    //与版本二的实现相同
});

2.4 第四版在mapTopair的过程中直接赋极小值

第三版实现性能得到了较大的提高，但是还不够好，因为其执行过程使用的转换过多。

下面的第四版实现为了减少转换的使用，就恢复到第一版的实现中。先对夜出时间的逻辑判断增强；另外，由于想到第一版的实现中最后的filter算子中需要value值为大于0才算夜出顾客，所以在mapToPair的过程中，每当得到白天出现的顾客A时就将它的value值设为一个较小的值，在后面的reduceByKey算子的执行过程中将晚上又再次出现的A的value值给抹平。那么最后的filter算子就能将白天出现过的顾客去除掉了！

下面就是具体实现，只是将版本一中的
return new Tuple2<String, Integer>(String.format("%s,%s",userName, strAry[0]), 0);
修改成
return new Tuple2<String, Integer>(String.format("%s,%s",userName,strAry[0]), -10000);
这样实现有些取巧，但是改动比较小，在性能上表现也可以在10秒内。

JavaPairRDD<String, Integer> pairRdd = df.toJavaRDD().mapToPair(new PairFunction<Row, String, Integer>() {
    @Override
    public Tuple2<String, Integer> call(Row row) throws Exception {
        String userName = row.getString(0);
        String[] strAry = Strsplit.splitByWholeSeparator(row.getString(1), " ", -1, true);
        if(nightTimeS.compareTo(nightTimeE) < 0){
            if(strAry[1].compareTo(nightTimeS) > 0 && strAry[1].compareTo(nightTimeE) < 0){
                if(strAry[1].compareTo("24:00:00") < 0 && strAry[1].compareTo("12:00:00") > 0){
                    return new Tuple2<String, Integer>(String.format("%s,%s",userName,strAry[0]), 1);
                }else{
                    Date preDate = DateUtil.parse(strAry[0], "yyyy-MM-dd");
                    String preDate_s = DateUtil.format(DateUtil.addDays(preDate, -1), "yyyy-MM-dd");
                    return new Tuple2<String, Integer>(String.format("%s,%s",userName,preDate_s), 1);
                }
            }else{
                // 白天出现的顾客的value设置为-10000
                return new Tuple2<String, Integer>(String.format("%s,%s",userName,strAry[0]), -10000);
            }
        }else{
            if(strAry[1].compareTo(nightTimeS) > 0){
                return new Tuple2<String, Integer>(String.format("%s,%s",userName,strAry[0]), 1);
            } else if(strAry[1].compareTo(nightTimeE) < 0){
                Date preDate = DateUtil.parse(strAry[0], "yyyy-MM-dd");
                String preDate_s = DateUtil.format(DateUtil.addDays(preDate, -1), "yyyy-MM-dd");
                return new Tuple2<String, Integer>(String.format("%s,%s",userName,preDate_s), 1);
            } else {
                // 白天出现的顾客的value设置为-10000
                return new Tuple2<String, Integer>(String.format("%s,%s",userName,strAry[0]), -10000);
            }
        }
    }
});

3 总结

通过多版本的Spark分析服务的改进，发现对Spark基础算子的正确使用至关重要。选择了合适的算子再配以合适的实现逻辑才能得到性能不错的Spark作业。

stillcoolme