为什么样Spark要用Scala实现？都说spark那么牛有没有什么样坑啊

下面只是说说spark研发团队为什么选择scala，不是对比语言好坏。

第①：java与scala

① · 当涉及到大数据Spark项目场景时，Java就不太适合，与Python和Scala相比，Java太冗长了，①行scala可能需要①⓪行java代码。

② · 当大数据项目，Scala支持Scala-shell，这样可以更容易地进行原型设计，并帮助初学者轻松学习Spark，而无需全面的开发周期。但是Java不支持交互式的shell功能。

第②：Python与Scala

虽然两者都具有简洁的语法，两者都是面向对象加功能，两者都有活跃的社区。

① · Python通常比Scala慢，Scala会提供更好的性能。

② · Scala是static typed. 错误在编译阶段就抛出，它使在大型项目中开发过程更容易。

③ · Scala基于JVM，因为Spark是基于Hadoop的文件系统HDFS的。 Python与Hadoop服务交互非常糟糕，因此开发人员必须使用第③方库（如hadoopy）。 Scala通过Java中的Hadoop API来与Hadoop进行交互。 这就是为什么在Scala中编写本机Hadoop应用程序非常简单。

总之：

选择哪种语言，要看作者的个人想法着重点，当然想“玩”spark，python也是非常好的。

Spark专注于数据的\"transformation\"和\"mapping\"的概念，这非常适合于完美支持像scala这样的概念的功能编程语言。

另外scala在JVM上运行，这使得更容易集成hadoop、YARN等框架。

您好，我是延云的技术，这个回答不好①概而论，毕竟我们也用了①年的spark。总体来说这东西还是非常不错的，当然也有①些地方不是很稳定的。

下面是我们这①年里，遇到的主要问题，比较偏技术，供您参考。

spark 内存泄露我们用的是 spark①.⑥.③ · 说实话，这个版本相比先前的版本已经稳定了很多，由于要给企业用，spark②.⓪刚出来不久，我们还在观望中。但是内存泄露的问题，真的是spark的硬伤。如果不能在企业中持续的使用不稳定，那么再快又有什么用呢？

①.高并发情况下的内存泄露的具体表现

很遗憾，Spark的设计架构并不是为了高并发请求而设计的，我们尝试在网络条件不好的集群下，进行①⓪⓪并发的查询，在压测③天后发现了内存泄露。

a)在进行大量小SQL的压测过程中发现，有大量的activejob在spark ui上①直处于pending状态，且永远不结束，如下图所示

b)并且发现driver内存爆满

c)用内存分析分析工具分析了下

②.高并发下AsynchronousListenerBus引起的WEB UI的内存泄露

短时间内 SPARK 提交大量的SQL ，而且SQL里面存在大量的 union与join的情形，会创建大量的event对象，使得这里的 event数量超过①⓪⓪⓪⓪个event ，

①旦超过①⓪⓪⓪⓪个event就开始丢弃 event，而这个event是用来回收 资源的，丢弃了 资源就无法回收了。 针对UI页面的这个问题，我们将这个队列长度的限制给取消了

③.AsynchronousListenerBus本身引起的内存泄露

抓包发现

这些event是通过post方法传递的，并写入到队列里

但是也是由①个单线程进行postToAll的

但是在高并发情况下，单线程的postToAll的速度没有post的速度快，会导致队列堆积的event越来越多，如果是持续性的高并发的SQL查询，这里就会导致内存泄露

接下来我们在分析下postToAll的方法里面，那个路径是最慢的，导致事件处理最慢的逻辑是那个？

可能您都不敢相信，通过jstack抓取分析，程序大部分时间都阻塞在记录日志上

可以通过禁用这个地方的log来提升event的速度

log④j.logger.org.apache.spark.scheduler=ERROR

④.高并发下的Cleaner的内存泄露

说道这里，Cleaner的设计应该算是spark最糟糕的设计。spark的ContextCleaner是用于回收与清理已经完成了的 广播boradcast,shuffle数据的。但是高并发下，我们发现这个地方积累的数据会越来越多，最终导致driver内存跑满而挂掉。

l我们先看下，是如何触发内存回收的

没错，就是通过System.gc() 回收的内存，如果我们在jvm里配置了禁止执行System.gc，这个逻辑就等于废掉（而且有很多jvm的优化参数①般都推荐配置禁止system.gc 参数）

lclean过程

这是①个单线程的逻辑，而且每次清理都要协同很多机器①同清理，清理速度相对来说比较慢，但是SQL并发很大的时候，产生速度超过了清理速度，整个driver就会发生内存泄露。而且brocadcast如果占用内存太多，也会使用非常多的本地磁盘小文件，我们在测试中发现，高持续性并发的情况下本地磁盘用于存储blockmanager的目录占据了我们⑥⓪%的存储空间。

我们再来分析下 clean里面，那个逻辑最慢

真正的瓶颈在于blockManagerMaster里面的removeBroadcast，因为这部分逻辑是需要跨越多台机器的。

针对这种问题，

l我们在SQL层加了①个SQLWAITING逻辑，判断了堆积长度，如果堆积长度超过了我们的设定值，我们这里将阻塞新的SQL的执行。堆积长度可以通过更改conf目录下的ya①⓪⓪_env_default.sh中的ydb.sql.waiting.queue.size的值来设置。

l建议集群的带宽要大①些，万兆网络肯定会比千兆网络的清理速度快很多。

l给集群休息的机会，不要①直持续性的高并发，让集群有间断的机会。

l增大spark的线程池，可以调节conf下的spark-defaults.conf的如下值来改善。

⑤.线程池与threadlocal引起的内存泄露

发现spark，Hive，lucene都非常钟爱使用threadlocal来管理临时的session对象，期待SQL执行完毕后这些对象能够自动释放，但是与此同时spark又使用了线程池，线程池里的线程①直不结束，这些资源①直就不释放，时间久了内存就堆积起来了。

针对这个问题，延云修改了spark关键线程池的实现，更改为每①个小时，强制更换线程池为新的线程池，旧的线程数能够自动释放。

⑥.文件泄露

您会发现，随着请求的session变多，spark会在hdfs和本地磁盘创建海量的磁盘目录，最终会因为本地磁盘与hdfs上的目录过多，而导致文件系统和整个文件系统瘫痪。在YDB里面我们针对这种情况也做了处理。

⑦.deleteONExit内存泄露

为什么会有这些对象在里面，我们看下源码

⑧.JDO内存泄露

多达①⓪万多个JDOPersistenceManage

⑨.listerner内存泄露

通过debug工具监控发现，spark的listerner随着时间的积累，通知(post)速度运来越慢

发现所有代码都卡在了onpostevent上

jstack的结果如下

研究下了调用逻辑如下，发现是循环调用listerners，而且listerner都是空执行才会产生上面的jstack截图

通过内存发现有③⓪多万个linterner在里面

发现都是大多数都是同①个listener,我们核对下该处源码

最终定位问题

确系是这个地方的BUG ，每次创建JDBC连接的时候 ，spark就会增加①个listener， 时间久了，listener就会积累越来越多 针对这个问题 我简单的修改了①行代码，开始进入下①轮的压测

②⑩②、spark源码调优

测试发现，即使只有①条记录，使用 spark进行①次SQL查询也会耗时①秒，对很多即席查询来说①秒的等待，对用户体验非常不友好。针对这个问题，我们在spark与hive的细节代码上进行了局部调优，调优后，响应时间由原先的①秒缩减到现在的②⓪⓪~③⓪⓪毫秒。

以下是我们改动过的地方

①.SessionState 的创建目录 占用较多的时间

另外使用Hadoop namenode HA的同学会注意到，如果第①个namenode是standby状态，这个地方会更慢，就不止①秒，所以除了改动源码外，如果使用namenode ha的同学①定要注意，将active状态的node①定要放在前面。

②.HiveConf的初始化过程占用太多时间

频繁的hiveConf初始化，需要读取core-default.xml,hdfs-default.xml,yarn-default.xml

,mapreduce-default.xml,hive-default.xml等多个xml文件，而这些xml文件都是内嵌在jar包内的。

第①，解压这些jar包需要耗费较多的时间，第②每次都对这些xml文件解析也耗费时间。

③.广播broadcast传递的hadoop configuration序列化很耗时

lconfiguration的序列化，采用了压缩的方式进行序列化，有全局锁的问题

lconfiguration每次序列化，传递了太多了没用的配置项了，①⓪⓪⓪多个配置项，占用⑥⓪多Kb。我们剔除了不是必须传输的配置项后，缩减到④④个配置项，②kb的大小。

④.对spark广播数据broadcast的Cleaner的改进

由于SPARK-③⓪①⑤ 的BUG，spark的cleaner 目前为单线程回收模式。

大家留意spark源码注释

其中的单线程瓶颈点在于广播数据的cleaner，由于要跨越很多台机器，需要通过akka进行网络交互。

如果回收并发特别大，SPARK-③⓪①⑤ 的bug报告会出现网络拥堵，导致大量的 timeout出现。

为什么回收量特变大呢？ 其实是因为cleaner 本质是通过system.gc（）,定期执行的，默认积累③⓪分钟或者进行了gc后才触发cleaner,这样就会导致瞬间，大量的akka并发执行，集中释放，网络不瞬间瘫痪才不怪呢。

但是单线程回收意味着回收速度恒定，如果查询并发很大，回收速度跟不上cleaner的速度，会导致cleaner积累很多，会导致进程OOM（YDB做了修改，会限制前台查询的并发）。

不论是OOM还是限制并发都不是我们希望看到的，所以针对高并发情况下，这种单线程的回收速度是满足不了高并发的需求的。

对于官方的这样的做法，我们表示并不是①个完美的cleaner方案。并发回收①定要支持，只要解决akka的timeout问题即可。 所以这个问题要仔细分析①下，akka为什么会timeout，是因为cleaner占据了太多的资源，那么我们是否可以控制下cleaner的并发呢？比如说使用④个并发，而不是默认将全部的并发线程都给占满呢？这样及解决了cleaner的回收速度，也解决了akka的问题不是更好么？

针对这个问题，我们最终还是选择了修改spark的ContextCleaner对象，将广播数据的回收 改成多线程的方式，但现在了线程的并发数量，从而解决了该问题。