Spark项目故障总结

lishenghan

　　1.OOM问题，reduce端的缓冲大小，太大的话，吃撑了，一下过来很多数据，容易OOM，默认48，可以改小哦。spark.reducer.maxSizeInFlight，48---》24
　　
　　2.JVM-GC导致的shuffle文件拉取失败，shuffle file not found
　　spark.shuffle.io.maxRetries 3 第一个参数，意思就是说，shuffle文件拉取的时候，如果没有拉取到（拉取失败），最多或重试几次（会重新拉取几次文件），默认是3次。 spark.shuffle.io.retryWait 5s 第二个参数，意思就是说，每一次重试拉取文件的时间间隔，默认是5s钟。 默认情况下，假如说第一个stage的executor正在进行漫长的full gc。第二个stage的executor尝试去拉取文件，结果没有拉取到，默认情况下，会反复重试拉取3次，每次间隔是五秒钟。最多只会等待3 * 5s = 15s。如果15s内，没有拉取到shuffle file。就会报出shuffle file not found。
　　spark.shuffle.io.maxRetries 60
　　spark.shuffle.io.retryWait 60s
　　3.YARN队列资源不足导致的application直接失败  :
　　限制提交一次作业，用线程池确保一个线程执行任务。
　　采用如下方案：
　　1、在你的J2EE（我们这个项目里面，spark作业的运行，之前说过了，J2EE平台触发的，执行spark-submit脚本），限制，同时只能提交一个spark作业到yarn上去执行，确保一个spark作业的资源肯定是有的。
　　2、你应该采用一些简单的调度区分的方式，比如说，你有的spark作业可能是要长时间运行的，比如运行30分钟；有的spark作业，可能是短时间运行的，可能就运行2分钟。此时，都提交到一个队列上去，肯定不合适。很可能出现30分钟的作业卡住后面一大堆2分钟的作业。分队列，可以申请（跟你们的YARN、Hadoop运维的同学申请）。你自己给自己搞两个调度队列。每个队列的根据你要执行的作业的情况来设置。在你的J2EE程序里面，要判断，如果是长时间运行的作业，就干脆都提交到某一个固定的队列里面去把；如果是短时间运行的作业，就统一提交到另外一个队列里面去。这样，避免了长时间运行的作业，阻塞了短时间运行的作业。
　　3、你的队列里面，无论何时，只会有一个作业在里面运行。那么此时，就应该用我们之前讲过的性能调优的手段，去将每个队列能承载的最大的资源，分配给你的每一个spark作业，比如80个executor；6G的内存；3个cpu core。尽量让你的spark作业每一次运行，都达到最满的资源使用率，最快的速度，最好的性能；并行度，240个cpu core，720个task。
　　4、在J2EE中，通过线程池的方式（一个线程池对应一个资源队列），来实现上述我们说的方案。
　　4.解决yarn-cluster模式的JVM内存溢出无法执行问题，（PermGen 永久代内存溢出）
　　既然是JVM的PermGen永久代内存溢出，那么就是内存不够用。咱们呢，就给yarn-cluster模式下的，driver的PermGen多设置一些。 spark-submit脚本中，加入以下配置即可： --conf spark.driver.extraJavaOptions="-XX:PermSize=128M -XX:MaxPermSize=256M" 这个就设置了driver永久代的大小，默认是128M，最大是256M。那么，这样的话，就可以基本保证你的spark作业不会出现上述的yarn-cluster模式导致的永久代内存溢出的问题。
　　5.错误的持久化方式以及checkpoint的使用
　　checkpoint原理：
　　1、在代码中，用SparkContext，设置一个checkpoint目录，可以是一个容错文件系统的目录，比如hdfs；
　　2、在代码中，对需要进行checkpoint的rdd，执行RDD.checkpoint()；
　　3、RDDCheckpointData（spark内部的API），接管你的RDD，会标记为marked for checkpoint，准备进行checkpoint
　　4、你的job运行完之后，会调用一个finalRDD.doCheckpoint()方法，会顺着rdd lineage，回溯扫描，发现有标记为待checkpoint的rdd，就会进行二次标记，inProgressCheckpoint，正在接受checkpoint操作
　　5、job执行完之后，就会启动一个内部的新job，去将标记为inProgressCheckpoint的rdd的数据，都写入hdfs文件中。（备注，如果rdd之前cache过，会直接从缓存中获取数据，写入hdfs中；如果没有cache过，那么就会重新计算一遍这个rdd，再checkpoint）
　　6、将checkpoint过的rdd之前的依赖rdd，改成一个CheckpointRDD*，强制改变你的rdd的lineage。后面如果rdd的cache数据获取失败，直接会通过它的上游CheckpointRDD，去容错的文件系统，比如hdfs，中，获取checkpoint的数据。
　　
　　数据倾斜
　　
　　1、定位原因与出现问题的位置: 根据log去定位 出现数据倾斜的原因，基本只可能是因为发生了shuffle操作，在shuffle的过程中，出现了数据倾斜的问题。因为某个，或者某些key对应的数据，远远的高于其他的key。 1、你在自己的程序里面找找，哪些地方用了会产生shuffle的算子，groupByKey、countByKey、reduceByKey、join
　　2、看log log一般会报是在你的哪一行代码，导致了OOM异常；或者呢，看log，看看是执行到了第几个stage！！！ 我们这里不会去剖析stage的划分算法
　　方案一：聚合源数据
　　将导致数据倾斜的一些shuffle操作，比如，groupbykey,reducebykey 或者join等，提前到ETL过程进行处理，一般情况spark处理的源数据来之hive ETL的hive表。
　　方案二：过滤导致倾斜的key
　　从源头过滤掉啦
　　方案三：提高reduce Task的数量，调用方式时，传入numPartitions
　　治标不治本的意思，因为，它没有从根本上改变数据倾斜的本质和问题。不像第一个和第二个方案（直接避免了数据倾斜的发生）。原理没有改变，只是说，尽可能地去缓解和减轻shuffle reduce task的数据压力，以及数据倾斜的问题。
　　方案四：双重groupby 在数据量过多的key上做文章咯。把这个key做分解，打散。比如key是“hello”做个map操作，变成“hello1”“hello2”。
　　使用场景 （1）groupByKey （2）reduceByKey 比较适合使用这种方式；
　　第一轮聚合的时候，对key进行打散，将原先一样的key，变成不一样的key，相当于是将每个key分为多组； 先针对多个组，进行key的局部聚合；接着，再去除掉每个key的前缀，然后对所有的key，进行全局的聚合。 对groupByKey、reduceByKey造成的数据倾斜，有比较好的效果。
　　方案五：（针对join操作）reduce join转换为map join，适合在什么样的情况下，可以来使用？ 如果两个RDD要进行join，其中一个RDD是比较小的。一个RDD是100万数据，一个RDD是1万数据。（一个RDD是1亿数据，一个RDD是100万数据） 其中一个RDD必须是比较小的，broadcast出去那个小RDD的数据以后，就会在每个executor的block manager中都驻留一份。要确保你的内存足够存放那个小RDD中的数据 这种方式下，根本不会发生shuffle操作，肯定也不会发生数据倾斜；从根本上杜绝了join操作可能导致的数据倾斜的问题；
　　方案六：（针对join操作）sample采样倾斜key进行两次join：这种方案什么时候适合使用？ 优先对于join，肯定是希望能够采用上一讲讲的，reduce join转换map join。两个RDD数据都比较大，那么就不要那么搞了。 针对你的RDD的数据，你可以自己把它转换成一个中间表，或者是直接用countByKey()的方式，你可以看一下这个RDD各个key对应的数据量；此时如果你发现整个RDD就一个，或者少数几个key，是对应的数据量特别多；尽量建议，比如就是一个key对应的数据量特别多。 此时可以采用咱们的这种方案，单拉出来那个最多的key；单独进行join，尽可能地将key分散到各个task上去进行join操作。 什么时候不适用呢？ 如果一个RDD中，导致数据倾斜的key，特别多；那么此时，最好还是不要这样了；还是使用我们最后一个方案，终极的join数据倾斜的解决方案。
　　方案七：使用随机数以及扩容表进行join：两个rdd都很大的jion操作，key进行打散，扩容10倍，在join