Spark优化一则 - 减少Shuffle
看了Spark Summit 2014的A Deeper Understanding of Spark Internals,视频(要科学上网)详细讲解了Spark的工作原理,Slides的45页给原始算法和优化算法。
破砂锅用自己3节点的Spark集群试验了这个优化算法,并进一步找到更快的算法。测试数据是Sogou实验室的日志文件前10000000条数据。目标是对日志第2列数据,按照第一个字母合并,得到每个首字母有几条记录。
所有的方案都重新启动Spark shell,先用以下代码把日志第2列数据cache到内存里,Spark GUI显示cache有8个partition,约1GB内存。
val rdd = sc.textFile("hdfs://hadoop1:8000/input/SogouQ3.txt").map(_.split("\t")).map(_(1)) rdd.cache() rdd.count() // res1: Long = 10000000
Spark GUI
RDD Name |
Storage Level |
Cached Partitions |
Fraction Cached |
Size in Memory |
Size in Tachyon |
Size on Disk |
3 |
Memory Deserialized 1x Replicated |
8 |
100% |
1089.4 MB |
0.0 B |
0.0 B |
Slides原始方案
rdd.map(x => (x.charAt(0), x)).groupByKey().mapValues({x => x.toSet.size}).collect() // res2: Array[(Char, Int)] = Array((8,168189), (0,168338), (a,168228), (9,168018), (1,167647), (b,168404), (2,168731), (3,168206), (c,168991), (d,168095), (4,167523), (e,168179), (5,167967), (6,167907), (f,168174), (7,168718))
Spark stage GUI显示有关stage Id是1-2,累计耗时5s,产生140MB shuffle read和208MB shuffle write。
Stage Id |
Description |
Submitted |
Duration |
Tasks: Succeeded/Total |
Shuffle Read |
Shuffle Write |
1 |
collect at <console>:15 |
2014/09/03 20:51:58 |
3 s |
8/8 |
140.2 MB |
|
2 |
map at <console>:15 |
2014/09/03 20:51:55 |
2 s |
8/8 |
208.4 MB |
|
0 |
count at <console>:15 |
2014/09/03 20:51:46 |
8 s |
8/8 |
Slides优化方案
rdd.distinct(numPartitions = 6).map(x => (x.charAt(0), 1)).reduceByKey(_+_).collect() // res2: Array[(Char, Int)] = Array((6,167907), (0,168338), (f,168174), (7,168718), (a,168228), (1,167647), (8,168189), (b,168404), (2,168731), (9,168018), (3,168206), (c,168991), (d,168095), (4,167523), (e,168179), (5,167967))
Spark stage GUI显示有关stage Id是1-3,累计耗时4.2s,生成50MB shuffle read和75MB shuffle write。虽然多了1个stage,shuffle read/write比原始方案减少超过60%,从而速度加快16%。
Stage Id |
Description |
Submitted |
Duration |
Tasks: Succeeded/Total |
Shuffle Read |
Shuffle Write |
1 |
collect at <console>:15 |
2014/09/03 20:24:17 |
0.2 s |
6/6 |
4.9 KB |
|
2 |
reduceByKey at <console>:15 |
2014/09/03 20:24:15 |
2 s |
6/6 |
50.4 MB |
7.4 KB |
3 |
distinct at <console>:15 |
2014/09/03 20:24:13 |
2 s |
8/8 |
75.6 MB |
|
0 |
count at <console>:15 |
2014/09/03 20:23:55 |
7 s |
8/8 |
Zero Shuffle优化方案
既然减少shuffle可以加快速度,破砂锅想出以下的Zero Shuffle方案来。
rdd.map(x => (x.charAt(0), x)).countByKey() // res2: scala.collection.Map[Char,Long] = Map(e -> 623689, 2 -> 623914, 5 -> 619840, b -> 626111, 8 -> 620738, d -> 623515, 7 -> 620222, 1 -> 616184, 4 -> 616628, a -> 641623, c -> 630514, 6 -> 621346, f -> 624447, 0 -> 632735, 9 -> 637770, 3 -> 620724)
Spark stage GUI显示有关stage Id是1,累计耗时只有0.3s,没有shuffle read/write。这个方案有关的RDD只有narrow dependency,所以只有1个stage。
Stage Id |
Description |
Submitted |
Duration |
Tasks: Succeeded/Total |
Shuffle Read |
Shuffle Write |
1 |
countByKey at <console>:15 |
2014/09/03 20:45:02 |
0.3 s |
8/8 |
||
0 |
count at <console>:15 |
2014/09/03 20:44:32 |
8 s |
小结
比较3种方案
方案 |
Shuffle Read |
Shuffle Write |
Time |
Slides原始方案 |
140.2 MB |
208.4 MB |
5s |
Slides优化方案 |
50.4 MB |
75.6 MB |
4.2s |
Zero Shuffle优化方案 |
0 |
0 |
0.3s |
Spark的优化之一是尽可能减少shuffle从而大幅减少缓慢的网络传输。熟悉RDD的函数对Spark优化有很大帮助。
本文转自http://www.cnblogs.com/frankbadpot/p/3969119.html,所有权力归原作者所有。
Spark优化一则 - 减少Shuffle相关推荐
- Spark性能调优案例-多表join优化,减少shuffle
背景 A任务在凌晨1点到3点,平均耗时1h,且是核心公共任务,急需优化. 整体逻辑示意图: // 从tableA读取一次数据,放到临时表t1 DROP TABLE IF EXISTS temp.tmp ...
- spark笔记spark优化
基本概念(Basic Concepts) RDD - resillient distributed dataset 弹性分布式数据集 Operation - 作用于RDD的各种操作分为transfor ...
- Hive On Spark优化
1.Hive官方建议的Hive On Spark优化 mapreduce.input.fileinputformat.split.maxsize=750000000hive.vectorized.ex ...
- 高并发大流量专题---3、前端优化(减少HTTP请求次数)
高并发大流量专题---3.前端优化(减少HTTP请求次数) 一.总结 一句话总结: 图片地图:使用<map><area></area></map>标签. ...
- 输入框实时搜索优化,减少请求次数、取消无用请求
背景 用户在输入框输入内容的同时进行搜索,针对搜索实时搜索结果进行展示 问题 如果不做处理,只监听输入框的input事件或者键盘按下弹起事件,实时对文本内容进行搜索,往往因为网络波动问题可能会造成结果 ...
- spark union 会引起shuffle吗_Spark高性能Job
知乎导入md文件会失真,无语,将就着看吧!原博客地址为:Spark高性能Job 1.1 Job 遇到一个action算子就会提交一个job,常见的transformation算子以及Action算子: ...
- Spark技术内幕:Shuffle Read的整体流程
回忆一下,每个Stage的上边界,要么需要从外部存储读取数据,要么需要读取上一个Stage的输出:而下边界,要么是需要写入本地文件系统(需要Shuffle),以供childStage读取,要么是最后一 ...
- Spark优化之小文件是否需要合并?
我们知道,大部分Spark计算都是在内存中完成的,所以Spark的瓶颈一般来自于集群(standalone, yarn, mesos, k8s)的资源紧张,CPU,网络带宽,内存.Spark的性能,想 ...
- Web项目性能优化之减少HTTP请求次数优化
作为项目经理的我,在近两个月内,时常被客户抱怨,我们的Web程序性能太差,经常宕机,即使服务器的配置再好,在业务请求量大的情况下也比较容易出现系统繁忙,线程堵塞,最后Web容器的HTTP线程池被慢慢耗 ...
最新文章
- 区块链公司Coingeek支持一系列BCH协议变更
- 也许开发需要的只是一份简单明了的表格
- oracle不足位数补0
- mysql 变量 视图_MySQL – 无法使用SET变量创建视图
- c语言 取随机字符串,C语言之获取32字节随机数的字符串
- 6.2创建Docker镜像文件
- 「代码随想录」关于多重背包,你该了解这些!
- linux内核之中断和异常
- 传智播客-刘意-java深入浅出精华版学习笔记Day07
- html网页中加载pdf,在HTML页面中嵌入本地PDF
- python爬虫技术作用_大数据爬虫技术有什么功能
- PSCAD中的频变参数线路模型(Frequency-Dependent (Phase) Line Model)
- ansys的kbc_ANSYS APDL 应用技巧
- QT 调试时出现 Unable to create a debugging engine.
- 【转】艺术设计、数字媒体、环艺、影视动画、摄影、广编专业…等…视频、教程、资讯、图库、作品汇总大全
- 虚拟机安装以及镜像激活码下载
- java.net.SocketException和错误:org.apache.ftpserver.FtpServerConfigurationException
- 基于粒子群优化算法的最小二乘支持向量机预测模型构建(MATLAB实现)
- 奢侈手机品牌,钱途路上的黄粱一梦
- 如何对人脸识别进行法律规制