spark 使用中会遇到的一些问题及解决思路 spark-shell命令行执行spark hql

朱雀 2022-06-04 07:30 657阅读 0赞

内存溢出问题

在Spark中使用hql方法执行hive语句时，由于其在查询过程中调用的是Hive的获取元数据信息、SQL解析，并且使用Cglib等进行序列化反序列化，中间可能产生较多的class文件，导致JVM中的持久代使用较多，如果配置不当，可能引起类似于如下的OOM问题：

1.  Exception in thread "Thread-2" java.lang.OutOfMemoryError: PermGen space

复制代码

原因是实际使用时，如果用的是JDK1.6版本， Server模式的持久代默认大小是64M，Client模式的持久代默认大小是32M，而Driver端进行SQL处理时，其持久代的使用可能会达到90M，导致OOM溢出，任务失败。

解决方法就是在Spark的conf目录中的 spark-defaults.conf里，增加对Driver的JVM配置，因为Driver才负责SQL的解析和元数据获取。配置如下：

1.  spark.driver.extraJavaOptions -XX:PermSize=128M -XX:MaxPermSize=256M

复制代码

但是，上述情况是在yarn-cluster模式下出现，yarn-client模式运行时倒是正常的，原来在$SPARK\_HOME/bin/spark-class文件中已经设置了持久代大小：

1.  JAVA\_OPTS="-XX:MaxPermSize=256m $OUR\_JAVA\_OPTS"  
    
2.

复制代码

当以yarn-client模式运行时，driver就运行在客户端的spark-submit进程中，其JVM参数是取的spark-class文件中的设置，所谓未出现持久代溢出现象。

总结一下Spark中各个角色的JVM参数设置：

(1)Driver的JVM参数：  
\-Xmx，-Xms，如果是yarn-client模式，则默认读取 [spark][]\-env文件中的SPARK\_DRIVER\_MEMORY值，-Xmx，-Xms值一样大小；如果是yarn-cluster模式，则读取的是spark-default.conf文件中的spark.driver.extraJavaOptions对应的JVM参数值。  
PermSize，如果是yarn-client模式，则是默认读取[spark][]\-class文件中的JAVA\_OPTS="-XX:MaxPermSize=256m $OUR\_JAVA\_OPTS"值；如果是yarn-cluster模式，读取的是spark-default.conf文件中的spark.driver.extraJavaOptions对应的JVM参数值。  
GC方式，如果是yarn-client模式，默认读取的是spark-class文件中的JAVA\_OPTS；如果是yarn-cluster模式，则读取的是spark-default.conf文件中的spark.driver.extraJavaOptions对应的参数值。  
以上值最后均可被spark-submit工具中的--driver-java-options参数覆盖。

(2)Executor的JVM参数：  
\-Xmx，-Xms，如果是yarn-client模式，则默认读取spark-env文件中的SPARK\_EXECUTOR\_MEMORY值，-Xmx，-Xms值一样大小；如果是yarn-cluster模式，则读取的是spark-default.conf文件中的spark.executor.extraJavaOptions对应的JVM参数值。  
PermSize，两种模式都是读取的是 [spark][]\-default.conf文件中的spark.executor.extraJavaOptions对应的JVM参数值。  
GC方式，两种模式都是读取的是spark-default.conf文件中的spark.executor.extraJavaOptions对应的JVM参数值。

(3)Executor数目及所占CPU个数  
如果是yarn-client模式，Executor数目由spark-env中的SPARK\_EXECUTOR\_INSTANCES指定，每个实例的数目由SPARK\_EXECUTOR\_CORES指定； 如果是yarn-cluster模式，Executor的数目由spark-submit工具的--num-executors参数指定，默认是2个实例，而每个Executor使用的CPU数目由--executor-cores指定，默认为1核。  
每个Executor运行时的信息可以通过yarn logs命令查看到，类似于如下：

1.  14/08/13 18:12:59 INFO org.apache.spark.Logging$class.logInfo(Logging.scala:58): Setting up executor with commands: List($JAVA\_HOME/bin/java, -server, -XX:OnOutOfMemoryError='kill %p', -Xms1024m -Xmx1024m , -XX:PermSize=256M -XX:MaxPermSize=256M -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintHeapAtGC -Xloggc:/tmp/[spark][]\_gc.log, -Djava.io.tmpdir=$PWD/tmp, -Dlog4j.configuration=log4j-spark-container.properties, org.apache.spark.executor.CoarseGrainedExecutorBackend, akka.tcp://spark@sparktest1:41606/user/CoarseGrainedScheduler, 1, sparktest2, 3, 1>, <LOG\_DIR>/stdout, 2>, <LOG\_DIR>/stderr)

复制代码

其中，akka.tcp://spark@sparktest1:41606/user/CoarseGrainedScheduler表示当前的Executor进程所在节点，后面的1表示Executor编号，sparktest2表示ApplicationMaster的host，接着的3表示当前Executor所占用的CPU数目。

8 序列化异常

在Spark上执行hive语句的时候，出现类似于如下的异常：

1.  org.apache.spark.SparkDriverExecutionException: Execution error  
    
2.      at org.apache.spark.scheduler.DAGScheduler.handleTaskCompletion(DAGScheduler.scala:849)  
    
3.      at org.apache.[spark][].scheduler.DAGSchedulerEventProcessActor$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1231)  
    
4.      at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)  
    
5.      at akka.actor.ActorCell.invoke(ActorCell.scala:456)  
    
6.      at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)  
    
7.      at akka.dispatch.Mailbox.run(Mailbox.scala:219)  
    
8.      at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)  
    
9.      at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)  
    
10.     at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)  
    
11.     at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)  
    
12.     at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)  
    
13. Caused by: java.lang.ClassCastException: scala.collection.mutable.HashSet cannot be cast to scala.collection.mutable.BitSet  
    
14.     at org.apache.spark.[sql][].execution.BroadcastNestedLoopJoin$anonfun$7.apply(joins.scala:336)  
    
15.     at org.apache.spark.rdd.RDD$anonfun$19.apply(RDD.scala:813)  
    
16.     at org.apache.spark.rdd.RDD$anonfun$19.apply(RDD.scala:810)  
    
17.     at org.apache.[spark][].scheduler.JobWaiter.taskSucceeded(JobWaiter.scala:56)  
    
18.     at org.apache.spark.scheduler.DAGScheduler.handleTaskCompletion(DAGScheduler.scala:845)

复制代码

排查其前后的日志，发现大都是序列化的东西：

1.  14/08/13 11:10:01 INFO org.apache.spark.Logging$class.logInfo(Logging.scala:58): Serialized task 8.0:3 as 20849 bytes in 0 ms  
    
2.  14/08/13 11:10:01 INFO org.apache.spark.Logging$class.logInfo(Logging.scala:58): Finished TID 813 in 25 ms on sparktest0 (progress: 3/200)

复制代码

而在 [spark][]\-default.conf中，事先设置了序列化方式为Kryo：

1.  [spark][].serializer org.apache.spark.serializer.KryoSerializer

复制代码

根据异常信息，可见是HashSet转为BitSet类型转换失败，Kryo把松散的HashSet转换为了紧凑的BitSet，把序列化方式注释掉之后，任务可以正常执行。难道Spark的Kryo序列化做的还不到位？ 此问题需要进一步跟踪。

9 Executor僵死问题

运行一个Spark任务，发现其运行速度远远慢于执行同样SQL语句的Hive的执行，甚至出现了OOM的错误，最后卡住达几小时！并且Executor进程在疯狂GC。

截取其一Task的OOM异常信息：

![235508fned0p3ni7d83plb.png][]

可以看到这是在序列化过程中发生的OOM。根据节点信息，找到对应的Executor进程，观察其Jstack信息：

1.  Thread 36169: (state = BLOCKED)  
    
2.  \- java.lang.Long.valueOf(long) @bci=27, line=557 (Compiled frame)  
    
3.  \- com.esotericsoftware.kryo.serializers.DefaultSerializers$LongSerializer.read(com.esotericsoftware.kryo.Kryo, com.esotericsoftware.kryo.io.Input, java.lang.Class) @bci=5, [line][]=113 (Compiled frame)  
    
4.  \- com.esotericsoftware.kryo.serializers.DefaultSerializers$LongSerializer.read(com.esotericsoftware.kryo.Kryo, com.esotericsoftware.kryo.io.Input, java.lang.Class) @bci=4, line=103 (Compiled frame)  
    
5.  \- com.esotericsoftware.kryo.Kryo.readClassAndObject(com.esotericsoftware.kryo.io.Input) @bci=158, line=732 (Compiled frame)  
    
6.  \- com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.read(com.esotericsoftware.kryo.Kryo, com.esotericsoftware.kryo.io.Input, java.lang.Class) @bci=158, [line][]=338 (Compiled frame)  
    
7.  \- com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.read(com.esotericsoftware.kryo.Kryo, com.esotericsoftware.kryo.io.Input, java.lang.Class) @bci=4, line=293 (Compiled frame)  
    
8.  \- com.esotericsoftware.kryo.Kryo.readObject(com.esotericsoftware.kryo.io.Input, java.lang.Class, com.esotericsoftware.kryo.Serializer) @bci=136, [line][]=651 (Compiled frame)  
    
9.  \- com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(com.esotericsoftware.kryo.io.Input, java.lang.Object) @bci=143, line=605 (Compiled frame)  
    
10. \- com.esotericsoftware.kryo.serializers.FieldSerializer.read(com.esotericsoftware.kryo.Kryo, com.esotericsoftware.kryo.io.Input, java.lang.Class) @bci=44, line=221 (Compiled frame)  
    
11. \- com.esotericsoftware.kryo.Kryo.readObject(com.esotericsoftware.kryo.io.Input, java.lang.Class, com.esotericsoftware.kryo.Serializer) @bci=136, [line][]=651 (Compiled frame)  
    
12. \- com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(com.esotericsoftware.kryo.io.Input, java.lang.Object) @bci=143, line=605 (Compiled frame)  
    
13. \- com.esotericsoftware.kryo.serializers.FieldSerializer.read(com.esotericsoftware.kryo.Kryo, com.esotericsoftware.kryo.io.Input, java.lang.Class) @bci=44, line=221 (Compiled frame)  
    
14. \- com.esotericsoftware.kryo.Kryo.readClassAndObject(com.esotericsoftware.kryo.io.Input) @bci=158, [line][]=732 (Compiled frame)  
    
15. \- org.apache.spark.serializer.KryoDeserializationStream.readObject(scala.reflect.ClassTag) @bci=8, line=118 (Compiled frame)  
    
16. \- org.apache.spark.serializer.DeserializationStream$anon$1.getNext() @bci=10, line=125 (Compiled frame)  
    
17. \- org.apache.spark.util.NextIterator.hasNext() @bci=16, line=71 (Compiled frame)  
    
18. \- org.apache.[spark][].storage.BlockManager$LazyProxyIterator$1.hasNext() @bci=4, [line][]=1031 (Compiled frame)  
    
19. \- scala.collection.Iterator$anon$13.hasNext() @bci=4, line=371 (Compiled frame)  
    
20. \- org.apache.spark.util.CompletionIterator.hasNext() @bci=4, line=30 (Compiled frame)  
    
21. \- org.apache.spark.InterruptibleIterator.hasNext() @bci=22, line=39 (Compiled frame)  
    
22. \- scala.collection.Iterator$anon$11.hasNext() @bci=4, line=327 (Compiled frame)  
    
23. \- org.apache.[spark][].sql.execution.HashJoin$anonfun$execute$1.apply(scala.collection.Iterator, scala.collection.Iterator) @bci=14, [line][]=77 (Compiled frame)  
    
24. \- org.apache.spark.[sql][].execution.HashJoin$anonfun$execute$1.apply(java.lang.Object, java.lang.Object) @bci=9, line=71 (Interpreted frame)  
    
25. \- org.apache.spark.rdd.ZippedPartitionsRDD2.compute(org.apache.spark.Partition, org.apache.spark.TaskContext) @bci=48, line=87 (Interpreted frame)  
    
26. \- org.apache.[spark][].rdd.RDD.computeOrReadCheckpoint(org.apache.spark.Partition, org.apache.spark.TaskContext) @bci=26, [line][]=262 (Interpreted frame)

复制代码

有大量的 BLOCKED线程，继续观察GC信息，发现大量的 FULL GC。

分析，在插入Hive表的时候，实际上需要写HDFS，在此过程的HashJoin时，伴随着大量的Shuffle写操作，JVM的新生代不断GC，Eden Space写满了就往Survivor Space写，同时超过一定大小的数据会直接写到老生代，当新生代写满了之后，也会把老的数据搞到老生代，如果老生代空间不足了，就触发FULL GC，还是空间不够，那就OOM错误了，此时线程被Blocked，导致整个Executor处理数据的进程被卡住。

当处理大数据的时候，如果JVM配置不当就容易引起上述问题。解决的方法就是增大Executor的使用内存，合理配置新生代和老生代的大小，可以将老生代的空间适当的调大点

# spark-shell命令行执行spark hql #

前面已经有篇文章介绍如何编译包含hive的spark-assembly.jar了,不清楚的可以翻看一下前面的文章。

cloudera manager装好的spark,直接执行spark-shell进入命令行后，写入如下语句：

val hiveContext = **new** org.apache.spark.sql.hive.HiveContext(sc)

你会发现没法执行通过，因为cm装的原生的spark是不支持spark hql的，我们需要手动进行一些调整：

第一步，将编译好的包含hive的JAR包上传到hdfs上配置的默认的spark的sharelib目录：/user/spark/share/lib

![Center][]

第二步：在你要运行spark-shell脚本的节点上的/opt/cloudera/parcels/CDH-5.3.0-1.cdh5.3.0.p0.30/lib/spark/lib/目录下面，下载这个jar到这个目录：hadoop fs -get hdfs://n1:8020/user/spark/share/lib/spark-assembly-with-hive-maven.jar（具体路径替换成你自己的）。然后这个目录下面原来会有个软链接spark-assembly.jar指向的是spark-assembly-1.2.0-cdh5.3.0-hadoop2.5.0-cdh5.3.0.jar，我们把这个软链接删除掉重新创建一个同名的软链接：ln -s spark-assembly-with-hive-maven.jar spark-assembly.jar，指向我们刚下载下来的那个JAR包，这个JAR包会在启动spark-shell脚本时装载到driver program的classpath中去的，sparkContext也是在driver中创建出来的，所以需要将我们编译的JAR包替换掉原来的spark-assembly.jar包，这样在启动spark-shell的时候，包含hive的spark-assembly就被装载到classpath中去了。

第三步：在/opt/cloudera/parcels/CDH/lib/spark/conf/目录下面创建一个hive-site.xml。/opt/cloudera/parcels/CDH/lib/spark/conf目录是默认的spark的配置目录，当然你可以修改默认配置目录的位置。hive-site.xml内容如下：

\[html\]  [view plain][]  [copy][view plain]

1.  <?xml version="1.0" encoding="UTF-8"?>  
2.    
3.    
4.  <configuration>  
5.    <property>  
6.      <name>hive.metastore.local</name>  
7.      <value>false</value>  
8.    </property>  
9.    <property>  
10.     <name>hive.metastore.uris</name>  
11.     <value>thrift://n1:9083</value>  
12.   </property>  
13.   <property>  
14.     <name>hive.metastore.client.socket.timeout</name>  
15.     <value>300</value>  
16.   </property>  
17.   <property>  
18.     <name>hive.metastore.warehouse.dir</name>  
19.     <value>/user/hive/warehouse</value>  
20.   </property>  
21. </configuration>

这个应该大家都懂的，总要让spark找到hive的元数据在哪吧，于是就有了上面一些配置。

第四步：修改/opt/cloudera/parcels/CDH/lib/spark/conf/spark-defaults.conf，添加一个属性：spark.yarn.jar=hdfs://n1:8020/user/spark/share/lib/spark-assembly-with-hive-maven.jar。这个是让每个executor下载到本地然后装载到自己的classpath下面去的，主要是用在yarn-cluster模式。local模式由于driver和executor是同一个进程所以没关系。

以上完事之后，运行spark-shell,再输入:

val hiveContext = **new** org.apache.spark.sql.hive.HiveContext(sc)

应该就没问题了。我们再执行一个语句验证一下是不是连接的我们指定的hive元数据库：

hiveContext.sql("show tables").take(10)   //取前十个表看看

最后要重点说明一下这里的第二步第三步和第四步，如果是yarn-cluster模式的话，应该替换掉集群所有节点的spark-assembly.jar集群所有节点的spark conf目录都需要添加hive-site.xml，每个节点spark-defaults.conf都需要添加spark.yarn.jar=hdfs://n1:8020/user/spark/share/lib/spark-assembly-with-hive-maven.jar。可以写个shell脚本来替换，不然手动一个一个节点去替换也是蛮累的。

值得欣喜的是，CDH5.9.x版本改善许多啊，只要配置一个地方就支持spark on hive了，把自己打包好的assembly jar放到hdfs上，配置一下spark jar location为hdfs上的assembly jar包：

![Center 1][]

[spark]: http://cpro.baidu.com/cpro/ui/uijs.php?c=news&cf=1001&ch=0&di=128&fv=16&jk=79c841810a573032&k=spark&k0=spark&kdi0=0&luki=5&n=10&p=baidu&q=92051019_cpr&rb=0&rs=1&seller_id=1&sid=3230570a8141c879&ssp2=1&stid=0&t=tpclicked3_hc&tu=u1946583&u=http%3A%2F%2Fwww.aboutyun.com%2Fthread-9425-1-1.html&urlid=0
[sql]: http://cpro.baidu.com/cpro/ui/uijs.php?c=news&cf=1001&ch=0&di=128&fv=16&jk=79c841810a573032&k=sql&k0=sql&kdi0=0&luki=2&n=10&p=baidu&q=92051019_cpr&rb=0&rs=1&seller_id=1&sid=3230570a8141c879&ssp2=1&stid=0&t=tpclicked3_hc&tu=u1946583&u=http%3A%2F%2Fwww.aboutyun.com%2Fthread-9425-1-1.html&urlid=0
[235508fned0p3ni7d83plb.png]: /images/20220604/12fd5f7c6b70453e8318d59809e7ece1.png
[line]: http://cpro.baidu.com/cpro/ui/uijs.php?c=news&cf=1001&ch=0&di=128&fv=16&jk=79c841810a573032&k=line&k0=line&kdi0=0&luki=3&n=10&p=baidu&q=92051019_cpr&rb=0&rs=1&seller_id=1&sid=3230570a8141c879&ssp2=1&stid=0&t=tpclicked3_hc&tu=u1946583&u=http%3A%2F%2Fwww.aboutyun.com%2Fthread-9425-1-1.html&urlid=0
[Center]: /images/20220604/eefbe1e3f4614321a069f4f711606b0f.png
[view plain]: http://blog.csdn.net/xiao_jun_0820/article/details/44345779#
[Center 1]: /images/20220604/0c719993fa0a495b875e3f5a7cf10701.png