Spark 的见解 & 优化 (四)

优化

这里调优主要针对于开发相关的调优见解，参数相关的调优在次不做赘述。

一：广播频繁使用的变量
在 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));

调优相关的官方资料：http://spark.apache.org/docs/latest/tuning.html

Spark 的见解 & 优化 (四)

相关帖子

(运维篇)- 使用 docker 搭建 hadoop-hive-spark 集群 (一)

python 从 0 编写 spark 程序

【翻译】Spark 的分区机制的应用及 PageRank 算法的实现

Spark SQL 操作 hive 过程 rename 过程时间长

Spark Streaming 实时统计数据（累加器的应用）

记一次数据类型不同导致的 sql join 异常

使用 spark 从 kafka 消费数据写入 hive 动态分区表（二）

欢迎来到这里！

近期热议

推荐标签标签

最新标签

Spark 的见解 & 优化 (四)

相关帖子

(运维篇)- 使用 docker 搭建 hadoop-hive-spark 集群 (一)

python 从 0 编写 spark 程序

【翻译】Spark 的分区机制的应用及 PageRank 算法的实现

Spark SQL 操作 hive 过程 rename 过程时间长

Spark Streaming 实时统计数据（累加器的应用）

记一次数据类型不同导致的 sql join 异常

使用 spark 从 kafka 消费数据写入 hive 动态分区表（二）

欢迎来到这里！

近期热议

推荐标签 标签

最新标签

推荐标签标签