一、数据本地化
1、背景
数据本地化对于Spark Job性能有着巨大的影响。如果数据以及要计算它的代码是在一起的,那么性能当然会非常高。但是,如果数据和计算它的代码是分开的, 那么其中之一必须到另外一方的机器上。通常来说,移动代码到其他节点,会比移动数据到代码所在的节点上去,速度要快得多,因为代码比较小。Spark也正是 基于这个数据本地化的原则来构建task调度算法的。 数据本地化,指的是,数据离计算它的代码有多近。基于数据距离代码的距离,有几种数据本地化级别: 1、PROCESS_LOCAL:数据和计算它的代码在同一个JVM进程中。 2、NODE_LOCAL:数据和计算它的代码在一个节点上,但是不在一个进程中,比如在不同的executor进程中,或者是数据在HDFS文件的block中。 3、NO_PREF:数据从哪里过来,性能都是一样的。 4、RACK_LOCAL:数据和计算它的代码在一个机架上。 5、ANY:数据可能在任意地方,比如其他网络环境内,或者其他机架上。
2、图解
Task要处理的partition的数据,在某一个Executor中,TaskScheduler首先会尽量用最好的本地化级别去启动task,也就是说,
会尽量在哪个包含了要处理的partition的executor中,去启动task;
此时,Executor已经再执行好几个task了,没有空闲资源来执行这个task;
默认情况下,spark会等待一会,等待Executor什么时候可以空闲出一个cpu core,从而来启动这个task,让它实现最好的本地化级别;
但是如果等待了一会(时间是可以调优的,通过参数设置),发现始终没有等到Executor的core释放,那么会放大一个级别,去尝试启动这个task;
如果这个rdd之前持久化过,task会去调用RDD的iterator()方法,然后通过executor关联的BlockManager,来尝试获取数据,BlockManager底层,
首先尝试从getLocal()在本地找数据,如果没有找到的话,那么用getRemote(),通过BlockTransferService,链接到有数据的BlockManager,来获取数据;
如果没有持久化过,那么就computerOrReadCheckpoint();
如果还是不能启动,继续放大级别;
3、数据本地化优化
Spark倾向于使用最好的本地化级别来调度task,但是这是不可能的。如果没有任何未处理的数据在空闲的executor上,那么Spark就会放低本地化级别。
这时有两个选择:第一,等待,直到executor上的cpu释放出来,那么就分配task过去;第二,立即在任意一个executor上启动一个task。
Spark默认会等待一会儿,来期望task要处理的数据所在的节点上的executor空闲出一个cpu,从而将task分配过去。只要超过了时间,
那么Spark就会将task分配到其他任意一个空闲的executor上。
可以设置参数,spark.locality系列参数,来调节Spark等待task可以进行数据本地化的时间。
spark.locality.wait(3000毫秒)、spark.locality.wait.node、spark.locality.wait.process、spark.locality.wait.rack。