优化
这里调优主要针对于开发相关的调优见解,参数相关的调优在次不做赘述。
一:广播频繁使用的变量
在 spark 运算中如果有利用 master 表的数据(只读)且数量较大的场景,那么这种场景利用广播该变量来提升性能。
不使用广播变量的场景:
那么 spark 就会把该数据以网络传输给各个 task,然后每个 task 都有该数据的副本,task 越多,对该数据副本的网络传输也越多,相对应的会加大网络的损耗。副本过多的话也会加大内存的损耗。
使用广播变量的场景:
那么 spark 就会把该数据的副本以广播的形式缓存到各个节点上,每个节点的所有 task 执行运算时共享该副本,因为每个节点只驻留一份副本,所有内存开销比不用广播变量的要小许多,而且 spark 使用有效的广播算法来分配广播变量,以降低通信成本。
SparkConf conf = new SparkConf();
conf.setMaster("local[1]");
conf.setAppName("test");
JavaSparkContext jc = new JavaSparkContext(conf);
Set masters = new HashSet<>();
masters.add(1);
masters.add(2);
masters.add(3);
List list = new ArrayList<>();
list.add(1);
list.add(2);
list.add(3);
list.add(4);
// 广播变量
Broadcast<Set<Integer>> broadcastVar = jc.broadcast(masters);
jc.parallelize(list).filter(x->broadcastVar.value().contains(x)?false:true).foreach(x-> System.out.println(x));
结果:
4
二:mapPartitions 替代 map,foreachPartition 替代 foreach
不带 partitions 是对应的函数进行转换/遍历所有数据,带 partitions 是对应的函数进行转换/遍历一个分区的数据,优缺点主要有以下几点:
1)带 partitions 的增加了并行度,对性能有一定的提升
2)如果要遍历数据往数据库插入数据,每一条数据都执行(创建连接-> 插入数据-> 释放连接)这样的流程的话,性能是十分低下的,而一个分区的数据只使用一个连接,然后批量插入这个分区的数据,最后再释放连接,这种性能上的提升是显著的
SparkConf conf = new SparkConf();
conf.setMaster("local[1]");
conf.setAppName("test");
JavaSparkContext jc = new JavaSparkContext(conf);
List<String> list = new ArrayList<>();
list.add("sql1");
list.add("sql2");
list.add("sql3");
list.add("sql4");
jc.parallelize(list).foreachPartition(x->{
// 获取连接
Connection connection = getConn();
x.forEachRemaining(x1->{
connection.prepareStatement(x1);
});
// 释放连接
connection.close();
});
三:先使用 map 跟 filter 操作对数据结构进行优化跟缩减
1)数据源最终都会变成一个广范的且有 schema 的对象,如果是计数/去重相关的操作,如果值是无关紧要的,就缩减为 0/1 等,减少内存占用。
2)做具体的业务之前先看一下是不是所有字段/属性都是必须的,如果有 10 个字段,你只取其中的 3 个字段的话,就尽量把多余的字段/属性给过滤掉。
3)上面 2 步在层次嵌套较深且数据量大的运算中有很大提升,因为不但减少了内存占用,Shuffle 传输的性能也有提高
SparkConf conf = new SparkConf();
conf.setMaster("local[1]");
conf.setAppName("test");
JavaSparkContext jc = new JavaSparkContext(conf);
List<Map<String,Object>> list = new ArrayList<>();
Map<String,Object> item = new HashMap<>();
item.put("a","a1");
item.put("b","b1");
item.put("c","c1");
item.put("d","d1");
item.put("e","e1");
list.add(item);
item = new HashMap<>();
item.put("a","a11");
item.put("b","b11");
item.put("c","c11");
item.put("d","d11");
item.put("e","e11");
item.put("f","e11");
list.add(item);
item = new HashMap<>();
item.put("b","b21");
item.put("d","d21");
item.put("e","e21");
list.add(item);
// 统计key=a出现的个数
jc.parallelize(list)
// 把map的每一条键值对变成多条记录
.flatMap(x->x.entrySet().iterator())
// 过滤a以外的key
.filter(x->"a".equals(x.getKey()))
// 因为统计key,value从string->int
.mapToPair(x->new Tuple2<>(x.getKey(),1))
// 统计
.reduceByKey((var1,var2)->var1+var2)
// 遍历
.foreachPartition(x-> x.forEachRemaining(x1->System.out.println(x1)));
结果:
(a,2)
四:使用 filter 后进行 coalesce
如果对 RDD 使用 filter 算子过渡掉较多的数据后,建议使用 coalesce 算子对分区进行缩减操作,使用 coalesce 缩减分区不会产生 Shuffle 操作,适用于数据量较少但分区数较多的场景
SparkConf conf = new SparkConf();
conf.setMaster("local[1]");
conf.setAppName("test");
JavaSparkContext jc = new JavaSparkContext(conf);
List<Map<String,Object>> list = new ArrayList<>();
Map<String,Object> item = new HashMap<>();
item.put("a","a1");
item.put("b","b1");
item.put("c","c1");
item.put("d","d1");
item.put("e","e1");
list.add(item);
item = new HashMap<>();
item.put("a","a11");
item.put("b","b11");
item.put("c","c11");
item.put("d","d11");
item.put("e","e11");
item.put("f","e11");
list.add(item);
item = new HashMap<>();
item.put("b","b21");
item.put("d","d21");
item.put("e","e21");
list.add(item);
// 统计key为a的出现的个数
JavaRDD, Object>> filter = jc.parallelize(list, 4)
// 把map的每一条键值对变成多条记录
.flatMap(x -> x.entrySet().iterator())
// 过滤a以外的key
.filter(x -> "a".equals(x.getKey()));
System.out.println("coalesce前分区数:"+filter.partitions().size());
// coalesce
JavaRDD, Object>> coalesce = filter.coalesce(1);
System.out.println("coalesce后分区数:"+coalesce.partitions().size());
结果:
coalesce前分区数:4
coalesce后分区数:1
五:使用 Kryo 进行序列化/反序列化
spark2.0.0 之前,默认使用的是 java 自带的序列化机制,同时 spark 也支持高性能的 kryo 序列化机制(官方说明比 java 的性能要快 10 倍),但是如果自定义的对象需要在 kryo 进行提前注册才能获得最佳性能。从 spark2.0.0 开始,spark 已经对 kryo 的 github 在这里,大家有兴趣可以自行研究一下。
// student类
public class Student {
private String name;
private int age;
private String sex;
public Student(String name, int age, String sex) {
this.name = name;
this.age = age;
this.sex = sex;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
public String getSex() {
return sex;
}
public void setSex(String sex) {
this.sex = sex;
}
}
// spark示例代码
SparkConf conf = new SparkConf();
conf.setMaster("local[1]");
conf.setAppName("test");
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
// conf.registerKryoClasses(new Class[]{Student.class});
JavaSparkContext jc = new JavaSparkContext(conf);
List<Student> list = new ArrayList<>();
for(int i=0;i<1000000;i++){
list.add(new Student("name_"+i,i,"male"));
}
JavaRDD students = jc.parallelize(list);
// 持久化
students.persist(StorageLevel.MEMORY_ONLY_SER());
System.out.println(students.count());
结果:
使用kryo序列化策略且没有提前注册自定义类的情况下:
INFO MemoryStore: Block rdd_0_0 stored as bytes in memory (estimated size 28.5 MB, free 883.8 MB)
使用kryo序列化策略且提前注册自定义类的情况下:
INFO MemoryStore: Block rdd_0_0 stored as bytes in memory (estimated size 20.9 MB, free 891.4 MB)
六:对多次使用的 RDD 进行持久化
spark 多次复用 RDD 的原理:每当你对这个 RDD 进行算子操作的时候,spark 都会从源头把这个 RDD 再算出来,然后再对这个 RDD 再进行算子操作,这种情况下,不对这个复用的 RDD 进行持久化操作的话,性能就很差。如果把这个 RDD 持久化到磁盘或者内存后,再针对于这个 RDD 进行算子操作的时候,直接从磁盘或者内存中提取这个 RDD,因为不用再从源头重新计算,这样效率就高了许多。五的示例已经有了持久化的示例,故不再写相关的示例代码。spark 对于持久化的策略有许多种,详细的请参考 org.apache.spark.storage.StorageLevel 这个类
七:尽量避免 Shuffle 算子
前一篇已经对 Shffle 操作,以及宽窄依赖做了比较详细的说明,所以尽量用非 Shuffle 的算子 + 广播变量等形式来完成业务,这样就可以尽量避免大规模的 IO 操作以及网络传输,可以大大减少性能开销。
八:使用 map-side 预聚合的 shuffle 操作
如果 Shuffle 算子是不可避免的,那么尽量使用 map-side 预聚合的算子。
所谓的 map-side 预聚合,就是每个节点都是对相同的 key 进行一次聚合操作,进行 map-side 预聚合后,每个节点都只会留下一个相同的 key,其他节点在拉取相同的 key 的时候,就会大大缩减拉取的数据,从而达到缩减减小 IO 跟网络开销的目的。
下面以 groupByKey 跟 reduceByKey 进行某个 key 的 sum 统计举例说明:
groupByKey:
reduceByKey:
groupByKey 示例代码:
SparkConf conf = new SparkConf();
conf.setMaster("local[1]");
conf.setAppName("test");
JavaSparkContext jc = new JavaSparkContext(conf);
List<Map<String,Integer>> list = new ArrayList<>();
Map<String,Integer> map = new HashMap<>();
map.put("yy",1);
map.put("jj",1);
map.put("ww",1);
list.add(map);
map = new HashMap<>();
map.put("yy",1);
list.add(map);
map = new HashMap<>();
map.put("yy",1);
map.put("jj",1);
list.add(map);
map = new HashMap<>();
map.put("yy",1);
map.put("jj",2);
list.add(map);
map = new HashMap<>();
map.put("jj",1);
map.put("ww",2);
list.add(map);
map = new HashMap<>();
map.put("ww",1);
list.add(map);
jc.parallelize(list)
// 遍历所有的key
.flatMap(x -> x.entrySet().iterator())
// 组成K,V键值对
.mapToPair(x -> new Tuple2<>(x.getKey(), x.getValue()))
// groupByKey
.groupByKey()
// sum
.mapValues(x->sum(x)).foreach(x-> System.out.println(x));
// sum函数
private static int sum(Iterable integers){
int sum = 0;
if(integers != null){
Iterator iterator = integers.iterator();
while (iterator.hasNext()){
sum += iterator.next();
}
}
return sum;
}
reduceByKey 示例代码:
SparkConf conf = new SparkConf();
conf.setMaster("local[1]");
conf.setAppName("test");
JavaSparkContext jc = new JavaSparkContext(conf);
List<Map<String,Integer>> list = new ArrayList<>();
Map<String,Integer> map = new HashMap<>();
map.put("yy",1);
map.put("jj",1);
map.put("ww",1);
list.add(map);
map = new HashMap<>();
map.put("yy",1);
list.add(map);
map = new HashMap<>();
map.put("yy",1);
map.put("jj",1);
list.add(map);
map = new HashMap<>();
map.put("yy",1);
map.put("jj",2);
list.add(map);
map = new HashMap<>();
map.put("jj",1);
map.put("ww",2);
list.add(map);
map = new HashMap<>();
map.put("ww",1);
list.add(map);
jc.parallelize(list)
// 遍历所有的key
.flatMap(x -> x.entrySet().iterator())
// 组成K,V键值对
.mapToPair(x -> new Tuple2<>(x.getKey(), x.getValue()))
// reduceByKey
.reduceByKey((var1,var2)->var1+var2).foreach(x-> System.out.println(x));
欢迎来到这里!
我们正在构建一个小众社区,大家在这里相互信任,以平等 • 自由 • 奔放的价值观进行分享交流。最终,希望大家能够找到与自己志同道合的伙伴,共同成长。
注册 关于