Spark+Cassandra优化
2015-06-04 16:36
239 查看
问题1:reduce task数目不合适
解决方案:
需要根据实际情况调整默认配置,调整方式是修改参数spark.default.parallelism。通常的,reduce数目设置为core数目的2-3倍。数量太大,造成很多小任务,增加启动任务的开销;数目太小,任务运行缓慢。所以要合理修改reduce的task数目即spark.default.parallelism
问题2:shuffle磁盘IO时间长
解决方案:
设置spark.local.dir为多个磁盘,并设置磁盘的IO速度快的磁盘,通过增加IO来优化shuffle性能;
问题3:map|reduce数量大,造成shuffle小文件数目多
解决方案:
通过设置spark.shuffle.consolidateFiles为true,来合并shuffle中间文件,此时文件数为reduce tasks数目;
问题4:序列化时间长、结果大
解决方案:
spark默认使用JDK 自带的ObjectOutputStream,这种方式产生的结果大、CPU处理时间长,可以通过设置spark.serializer为org.apache.spark.serializer.KeyoSerializer。
另外如果结果已经很大,那就最好使用广播变量方式了,结果你懂得。
问题5:单条记录消耗大
解决方案:
使用mapPartition替换map,mapPartition是对每个Partition进行计算,而map是对partition中的每条记录进行计算;
问题6 : collect输出大量结果时速度慢
解决方案:
collect源码中是把所有的结果以一个Array的方式放在内存中,可以直接输出到分布式的文件系统,然后查看文件系统中的内容;
问题7: 任务执行速度倾斜
解决方案:
如果数据倾斜,一般是partition key取得不好,可以考虑其他的并行处理方式,并在中间加上aggregation操作;如果是Worker倾斜,例如在某些Worker上的executor执行缓慢,可以通过设置spark.speculation=true 把那些持续慢的节点去掉;
问题8: 通过多步骤的RDD操作后有很多空任务或者小任务产生
解决方案:
使用coalesce或者repartition去减少RDD中partition数量;
问题9:Spark Streaming吞吐量不高
可以设置spark.streaming.concurrentJobs
问题10:Spark Streaming 运行速度突然下降了,经常会有任务延迟和阻塞
解决方案:
这是因为我们设置job启动interval时间间隔太短了,导致每次job在指定时间无法正常执行完成,换句话说就是创建的windows窗口时间间隔太密集了;
RDD函数使用的一些问题
collect
如果数据集特别大,不要贸然使用collect,因为collect会将计算结果统统的收集返回到driver节点,这样非常容易导致driver结点内存不足,程序退出
repartition
在所能提供的core数目不变的前提下,数据集的分区数目越大,意味着计算一轮所花的时间越多,因为中间的通讯成本较大,而数据集的分区越小,通信开销小而导致计算所花的时间越短,但数据分区越小意味着内存压力越大。
假设为每个spark application提供的最大core数目是32,那么将partition number设置为core number的两到三倍会比较合适,即parition number为64~96。
/tmp目录问题
由于Spark在计算的时候会将中间结果存储到/tmp目录,而目前linux又都支持tmpfs,其实说白了就是将/tmp目录挂载到内存当中。
那么这里就存在一个问题,中间结果过多导致/tmp目录写满而出现如下错误
解决办法就是针对tmp目录不启用tmpfs,修改/etc/fstab,如果是archlinux,仅修改/etc/fstab是不够的,还需要执行如下指令:
3.4 Cassandra的配置优化
3.4.1 表结构设计
Cassandra表结构设计的一个重要原则是先搞清楚要对存储的数据做哪些操作,然后才开始设计表结构。如:
只对表进行添加,查询操作
对表需要进行添加,修改,查询
对表进行添加和修改操作
一般来说,针对Cassandra中某张具体的表进行“添加,修改,查询”并不是一个好的选择,这当中会涉及到效率及一致性等诸多问题。
Cassandra比较适合于添加,查询这种操作模式。在这种模式下,需要先搞清楚要做哪些查询然后再来定义表结构。
加深对Cassandra中primary key及其变种的理解有利于设计出高效查询的表结构。
上述例子中primary key由(k,v)组成,其中k是partition key,而v是clustering columns,如果k相同,那么这些记录在物理存储上其实是存储在同一行中,即Cassandra中常会提及的wide rows.
有了这个基础之后,就可以进行范围查询了
Cassandra中针对二级索引是不支持范围查询的,一切的一切都在主键里打主意。
解决方案:
需要根据实际情况调整默认配置,调整方式是修改参数spark.default.parallelism。通常的,reduce数目设置为core数目的2-3倍。数量太大,造成很多小任务,增加启动任务的开销;数目太小,任务运行缓慢。所以要合理修改reduce的task数目即spark.default.parallelism
问题2:shuffle磁盘IO时间长
解决方案:
设置spark.local.dir为多个磁盘,并设置磁盘的IO速度快的磁盘,通过增加IO来优化shuffle性能;
问题3:map|reduce数量大,造成shuffle小文件数目多
解决方案:
通过设置spark.shuffle.consolidateFiles为true,来合并shuffle中间文件,此时文件数为reduce tasks数目;
问题4:序列化时间长、结果大
解决方案:
spark默认使用JDK 自带的ObjectOutputStream,这种方式产生的结果大、CPU处理时间长,可以通过设置spark.serializer为org.apache.spark.serializer.KeyoSerializer。
另外如果结果已经很大,那就最好使用广播变量方式了,结果你懂得。
问题5:单条记录消耗大
解决方案:
使用mapPartition替换map,mapPartition是对每个Partition进行计算,而map是对partition中的每条记录进行计算;
问题6 : collect输出大量结果时速度慢
解决方案:
collect源码中是把所有的结果以一个Array的方式放在内存中,可以直接输出到分布式的文件系统,然后查看文件系统中的内容;
问题7: 任务执行速度倾斜
解决方案:
如果数据倾斜,一般是partition key取得不好,可以考虑其他的并行处理方式,并在中间加上aggregation操作;如果是Worker倾斜,例如在某些Worker上的executor执行缓慢,可以通过设置spark.speculation=true 把那些持续慢的节点去掉;
问题8: 通过多步骤的RDD操作后有很多空任务或者小任务产生
解决方案:
使用coalesce或者repartition去减少RDD中partition数量;
问题9:Spark Streaming吞吐量不高
可以设置spark.streaming.concurrentJobs
问题10:Spark Streaming 运行速度突然下降了,经常会有任务延迟和阻塞
解决方案:
这是因为我们设置job启动interval时间间隔太短了,导致每次job在指定时间无法正常执行完成,换句话说就是创建的windows窗口时间间隔太密集了;
RDD函数使用的一些问题
collect
如果数据集特别大,不要贸然使用collect,因为collect会将计算结果统统的收集返回到driver节点,这样非常容易导致driver结点内存不足,程序退出
repartition
在所能提供的core数目不变的前提下,数据集的分区数目越大,意味着计算一轮所花的时间越多,因为中间的通讯成本较大,而数据集的分区越小,通信开销小而导致计算所花的时间越短,但数据分区越小意味着内存压力越大。
假设为每个spark application提供的最大core数目是32,那么将partition number设置为core number的两到三倍会比较合适,即parition number为64~96。
/tmp目录问题
由于Spark在计算的时候会将中间结果存储到/tmp目录,而目前linux又都支持tmpfs,其实说白了就是将/tmp目录挂载到内存当中。
那么这里就存在一个问题,中间结果过多导致/tmp目录写满而出现如下错误
No Space Left on the device
解决办法就是针对tmp目录不启用tmpfs,修改/etc/fstab,如果是archlinux,仅修改/etc/fstab是不够的,还需要执行如下指令:
systemctl mask tmp.mount
3.4 Cassandra的配置优化
3.4.1 表结构设计
Cassandra表结构设计的一个重要原则是先搞清楚要对存储的数据做哪些操作,然后才开始设计表结构。如:
只对表进行添加,查询操作
对表需要进行添加,修改,查询
对表进行添加和修改操作
一般来说,针对Cassandra中某张具体的表进行“添加,修改,查询”并不是一个好的选择,这当中会涉及到效率及一致性等诸多问题。
Cassandra比较适合于添加,查询这种操作模式。在这种模式下,需要先搞清楚要做哪些查询然后再来定义表结构。
加深对Cassandra中primary key及其变种的理解有利于设计出高效查询的表结构。
create test ( k int, v int , primary key(k,v))
上述例子中primary key由(k,v)组成,其中k是partition key,而v是clustering columns,如果k相同,那么这些记录在物理存储上其实是存储在同一行中,即Cassandra中常会提及的wide rows.
有了这个基础之后,就可以进行范围查询了
select * from test where k = ? and v > ? and v < ?当然也可以对k进行范围查询,不过要加token才行,但一般这样的范围查询结果并不是我们想到的
select * from test where token(k) > ? and token(k) < ?
Cassandra中针对二级索引是不支持范围查询的,一切的一切都在主键里打主意。
相关文章推荐
- 中兴阅读:你的移动阅读解决专家,助纸媒们一臂之力
- 14.1 将计算机加入域
- MFC DLL PreTranslateMessage 导致的快捷键不响应的问题?
- POJ2406简单KMP
- poj2418map或者字典树
- SpringMVC 学习笔记(六) 数据绑定和JSR校验
- 004-流程控制和类型转换
- ccs平台 28335mcu 关于变量重复定义的解决方案
- CrashMonkey4Android
- Android报Caused by: android.content.res.Resources$NotFoundException: String resource ID #0x0 .解决办法
- 用户画像的一些相关信息链接
- [picture]saber
- oracle 11g sql developer安装后无法使用
- 树形列表成员- DevExpress.XtraTreeList.TreeList
- Linux下为PHP添加MongoDB扩展
- Hdu 1059 Dividing(dp)
- 死锁的原理
- 史上最牛最快速解决此windows副本不是正版的技巧
- 关于百度推送iOS开放技术文档
- 新浪微博开发平台地址 http://open.weibo.com/