原文链接：https://www.cnblogs.com/stillcoolme/p/10160397.html
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太过冗余，在Spark官网的Tuning Data Structures中就明确说过集合类型和包装类型的java对象占用了太多的额外空间，会降低执行效率，强烈不建议使用，并提出以下建议：
- 设计数据结构成数组类型和原始类型，而不是标准的Java或者Sacala集合
- 避免在数据结构里面嵌套大量的小对象。
- 考虑使用数值型的id或者enumeration对象而不是String类型的key，由于String类型会占用额外的字节。
- 如果使用小于32G的内存，可以设置JVM参数：-XX:+UseCompressedOops 使指针占用4bytes而不是8bytes，可以加这些配置在spark-env.sh文件中。
所以第三版的实现在这问题上做了改进。
通过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作业。