目录
1.系统架构
2.环境搭建
2.1本地环境下kafka批量导入数据
2.2 kafka-manager的安装与配置
3.1 Spark Streaming 性能调优(一): 解决并行度
3.2 Spark Streaming 性能调优(二): 解决task倾斜
前一篇文章解决了task数据倾斜的问题, 但是将代码提交放到集群环境上测试的时候却发现, 性能并没有多大改善, 但是通过Spark ui可以看到, 各个task分配到的数据量确实是比较平均的, 而奇怪的就在于, 这些数据量平均的task却都集中在某一两台机器上运行了.
task倾斜
上面我遇到的这种情况其实就task倾斜, 各个task的数据量很平均, 但都集中在某一两机器上, 导致这一两台机器性能被榨干, 其他机器在围观.
之后经过了几天理解Spark的task分配机制, 最后终于发现, 我目前遇到的task倾斜都是由于Spark的数据本地化特性导致的.(/(ㄒoㄒ)/~~)
数据本地化
数据本地化有3个级别:
- PROCESS_LOCAL: 该级别由参数spark.locality.wait.process指定
- NODE_LOCAL: 该级别由参数spark.locality.wait.node指定
- RACK_LOCAL: 该级别由参数spark.locality.wait.rack指定
上面三个参数, 如果不指定值的话, 则默认取值于参数spark.locality.wait的值, 该值默认为3s
Spark在task的调度过程中, 会优化把task分配给持有数据/缓存的同一个executor进程, 如果executor进程正在执行别的任务, 则会等待spark.locality.wait.process秒, 如果等待了这么多秒之后该executor还是不可用, 则数据本地化程度降低一级, 选择分配到统一节点的其他executor进程, 如果还是不可用, 则选择同一个集群的其他executor.
加入通过spark ui的executor页发现某个executor的数据input量特别大, 则极有可能会发生task倾斜.
解决办法
解决办法其实就是适当降低数据本地化参数的值, 具体的值是多少需要视实际情况.
由于我公司的集群可以开16个executor, 而发生task倾斜的时候每个task运行时间大概也就2~3s, 所以我设置了spark.locality.wait.process=200ms, spark.locality.wait.node=200ms, 其他用默认值.
修改配置参数后重新提交spark程序, 确实集群的资源终于能利用起来了, 原本要1分钟的窗口间隔, 要跑1min~1.5min, 现在只需30~40s就跑完了, 终于不会再出现延时严重的情况了!!!
((≧▽≦)/啦啦啦)
