当前位置:   article > 正文

Spark join问题_spark full join

spark full join
本篇文章是关于我在学习 Spark 过程中遇到的一些问题及总结,分为Spark优化、RDD join问题、遇到的问题、总结、参考资料几个部分。 

一:Spark优化 
1、设置序列化器为KryoSerializer 
  Spark默认使用的是Java序列化机制,但是Spark也支持使用Kryo序列化库,Kryo序列化机制比Java序列化机制性能高10倍左右 
这里写图片描述 
2、压缩机制 
  如果数据量很大,在序列化的同时可以考虑使用压缩,lzf的压缩效率要高很多, 当然如果使用了压缩,也会消耗CPU和内存资源 
这里写图片描述 
3、TDW 表写入禁止覆盖 
  最终写入TDW的RDD可能是多个,因此需要设置TDW表写入禁止覆盖,这样就可以避免使用RDD UNION操作 
这里写图片描述 
4、启用Spark推测机制 
这里写图片描述 
5、reduceByKey代替groupByKey 
  groupByKey会将所有的计算放在reduce阶段进行,会导致全量数据在节点间传输,而reduceByKey会在map端对本地数据进行聚合,之后将计算的结果进行shuffle,因此可以大量的减少shuffle的数据,减少网络IO,提高运行效率,在reduceByKey的函数中,还可以去实现某个字段的sum,max,min,count等操作 
6、map和mapPartitions 
  mapPartitions是针对Partitiion进行操作,那么操作中的很多对象和变量都可以复用,比如广播变量等 
  map是处理partitiion中的一条数据,因此mapPartition的效率要高一些,但是mapPartitions也存在缺陷,由于一次处理一个partitiion的数据,在内存不足的时候会因此OOM 
7、foreach和foreachPartitions 
  类似与map和mapPartitions的关系,前者是针对partition中的数据一条条进行处理,后者是针对一个partition进行处理,后者适合在和外部数据库交互操作时使用,比如MySQL,通过这种方法可以避免频繁的创建和销毁链接,还可以进行批处理,比如使用JDBC在mysql数据库中批量插入,同样也存在缺陷,会遇到OOM 
8、cache和persist 
  如果需要重复使用RDD,可以考虑使用缓存操作,cache是将RDD缓存到内存中,适合数据量比较小的RDD,对于persist而言,可以根据不同的业务场景选择不同的持久化级别。 
这里写图片描述

二:RDD join 问题 
  程序RDD 的join问题的一些总结:A表join B表,对B表进行更新问题,都可以把表转换成(k,v)的形式(k可以由多个字段拼接而成,v可以是一个对象,也可以是一个字段),最终问题转化为A(k,v) join B(k,v) on A.k = B.k 
1、full join 
这里写图片描述 
使用场景:A表和B表进行full join,并且A表比B表的数据量不在一个数量级,如果在一个数量级,则直接使用full out join算子执行 
方案: 
①把full out join转化为A表left out join B表 
  ②之后进行map操作,根据A对B进行更新 
  ③map操作中用HashSet记录A 和B交集的key 
  ④根据③中得到的HashSet对B进行filter操作,过滤掉已经left out join的记录,然后对B中剩下的记录进行foreach遍历更新 
  ⑤调用TDW接口把RDD分别写入文件中 
代码: 
这里写图片描述 
2、小表join大表 
使用场景: 
   一个小表A和一个大表B的连接操作,小表指文件足够小,可以加载到内存中,该算法可以将join算子执行在Map端,无需经历shuffe和reduce等阶段,因此效率很高 
方案: 
  ①把A(k,v)进行广播 
  ②在Map端对B进行过滤和更新 
  ③把更新后的RDD写入文件中 
代码: 
这里写图片描述
3、大表join大表 
使用场景: 
   当两个表数据量非常大,其中任何一个都不能够放到内存中,可以使用Spark的join算子,通过该算子实现reduce-side-join 
概念:RDD依赖关系 
在spark中如何表示RDD之间的依赖关系分为两类: 
①窄依赖:每个父RDD的分区都至多被一个子RDD的分区使用,即为OneToOneDependecies; 
②宽依赖:多个子RDD的分区依赖一个父RDD的分区,即为OneToManyDependecies。 
例如,map操作是一种窄依赖,而join操作是一种宽依赖(除非父RDD已经基于Hash策略被划分过了) 
这里写图片描述 
  图:宽依赖和窄依赖 
方案: 
   由上图可知,join分为宽依赖和窄依赖,如果RDD有相同的partitioner,那么将不会引起shuffle,因此我们可以对RDD进行Hash分区。分别对A和B用同一个函数进行Partition,比如按照首字母进行Partition,那么A和B都可以分成26个Partition,并且A1只需要和B1进行join,A1不需要和B剩下的25个Partition进行join,这样就大大的减少了join次数,最好的办法是对表进行分区,每次只取两个对应分区的数据进行join操作。具体的Hash Partition函数需要根据具体的应用场景实现,比如:如果key是URL,那么就可以根据域名进行分区。分区大小需要根据task-nums、num-executors以及executor-cores确定。 
这里写图片描述 
   图:Spark中CoGroupedRDD.Scala源码 
代码: 
这里写图片描述 
图:分区函数 
这里写图片描述 
图:对RDD进行分区 
4、RDD join引起 shuffle问题 
  请参考《Spark性能优化指南》系列,里边详细讲解了引起shuffle的原因,以及不通场景的解决办法,强烈推荐。 
   
三:遇到的问题 
   列出我在写Spark程序遇到的两个问题: 
1、RDD的API所引用的所有对象,都必须是可序列化的 
这里写图片描述
图:RDD中引用对象 
这里写图片描述
图:报错信息

   在RDD的API里所引用的在RDD的API里所引用的所有对象,都必须是可序列化的,因为RDD分布在多台机器是,代码和所引用的对象会序列化,然后复制到多台机器,所以凡是被引用的数据,都必须是可序列化的。否则会报java.lang.NotSerializableException: scala.util.Random 异常,解决办法就是把引用对象序列化 extends Serializable或者使用kryo序列化。 
在一个RDD的api里不可以引用另外一个RDD 
  SPARK-5063 in spark,Spark does not support nested RDDs or performing Spark actions inside of transformations; this usually leads to NullPointerExceptions (seeSPARK-718 as one example). The confusing NPE is one of the most common sources of Spark questions on StackOverflow: 
  上边英文大致意思是:Spark的transformation算子中不支持嵌套RDD,会导致空指针,如果其中一个RDD数据量不大,则可以用文章中提高的广播变量解决这个问题,如果数据量很大使用广播变量会导致OOM,那么就要从其他方面进行优化或者从业务逻辑进行出发。

四:总结 
  这一个月我的主要工作是把Sql转化成Spark程序,以及去不断的去优化,提高效率。在把Sql转化成Spark程序有很多可以优化的点,我选择了其中一个点进行了总结,总结的有不妥的地方,欢迎拍砖,一起交流。 
  时间过的很快,在这里已经实习一个月了,在导师dezhiliu和小组成员的帮助下 ,自己成长了很多,也学到了很多东西,不仅仅是Spark相关的。特别的感谢我的导师和我的小伙伴们。

五:参考资料 
[1]、Spark性能优化指南——基础篇 
  http://tech.meituan.com/spark-tuning-basic.html 
[2]、Spark性能优化指南——高级篇 
  http://tech.meituan.com/spark-tuning-pro.html 
[3]、Apache Spark探秘:实现Map-side Join和Reduce-side Join http://dongxicheng.org/framework-on-yarn/apache-spark-join-two-tables/ 
[4]、Spark排错与优化 
  http://blog.csdn.net/lsshlsw/article/details/49155087 
[5]、Spark官方文档 
  http://spark.apache.org/docs/latest/

1
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/爱喝兽奶帝天荒/article/detail/771928?site
推荐阅读
相关标签
  

闽ICP备14008679号