Spark性能调优（六）：数据本地性

1.什么是数据本地性？2.数据本地性有哪些策略？

        数据本地性指的是数据与执行的代码的远近程度。基于数据与执行的代码各自所在的位置，从近到远的本地性级别依次是：
        1.PROCESS_LOCAL：数据与执行的代码在同一个JVM内存中，数据已经缓存到执行代码所在的executor的内存中了，即读取缓存在本地节点的数据，这是效率最高的数据本地性级别；（该本地性级别和cache有关）
        2.NODE_LOCAL： 数据与执行的代码在同一个节点上，需要从执行代码所在的节点上的本地存储中读取出数据，即读取本地节点硬盘数据。比如，数据在同一个节点的hdfs上，或在同一个节点的另一个executor上。因为数据需要在进程间移动，该级别比PROCESS_LOCAL稍慢；
        3.NO_PREF：从任意地方访问该数据的速度都一样，没有具体的位置偏好；
        4.RACK_LOCAL：数据所在的server与执行代码的server在同一台机架上，一般需要通过一台交换机在网络上传输到执行代码所在的节点上；
        5.ANY：数据位于不在同一台机架的其他集群的节点上，即读取非本地节点数据。
数据本地性对程序的性能有很大的影响。如果数据和要执行的代码在一起的话，计算速度就很快；而当它们不在一起时，一个就必须移动到另一个所在的位置，而数据在网络上的传输会导致大量的延时和开销，毕竟磁盘 IO和网络IO都是耗时的操作。
        在大数据处理模式下，一般代码都比数据小, 所以序列化了的代码的移动比数据的移动更快，分布式计算系统的精粹就在于移动代码而非移动数据，即‘数据不动代码动’。数据本地性的原则就是尽量避免数据在网络上甚至磁盘上的传输，尽量将计算移到数据所在的节点上进行。spark的调度机制就是基于这一准则来进行调度的。
        Spark中任务的处理需要考虑数据本地性的场合，基本就两种：一是数据来源于外部数据源比如HDFS; 二是数据来源于RDD Cache（即由CacheManager从BlockManager中读取，或者Streaming数据源RDD）。其他情况下，不涉及shuffle操作的不构成划分Stage和Task的基准，也就不存在本地性问题；而如果是ShuffleRDD，由于其本地性始终为No Prefer，因此其实也不存在本地性问题。
        在理想的情况下，任务当然是分配在可以从本地读取数据的节点上时（同一个JVM内部或同一台物理机器内部）的运行时性能最佳。但是每个任务的执行速度无法准确估计，所以很难在事先获得全局最优的执行策略。当Spark应用程序得到一个计算资源的时候，如果没有可以满足最佳本地性需求的任务可以运行时，即任何空闲的executor上都没有尚未处理的数据时，计算框架通常有两种选择：1.一直等待直到待处理的数据所在的节点的cpu空闲下来，然后调度处理该批数据的task到该节点上，这样能更好的匹配任务的本地性；2.不进行等待，直接将处理该批数据的task调度到其他节点上执行，当然此时该批数据需要移动到相应的节点上, 这是退而求其次，运行一个本地性条件稍差一点的任务。
        Spark所遵循的是，调度作业时优先调度到有最好本地性的节点上去执行task，当较高的数据本地性级别不能满足时，spark退而求其次调度作业到次好的本地性节点。具体实施时，spark混合了这两种方案，即会首先等待一小会以期望忙碌的cpu空闲下来，如果这个等待的‘一小会’到了而cpu仍然忙碌的话，就会移动数据到有空闲cpu的节点去执行。这个需要等待的‘一小会’是可以通过参数spark.locality.wait, spark.locality.wait.process, spark.locality.wait.node和 spark.locality.wait.rack来配置的，这几个参数一起决定了Spark任务调度模块在得到待分配任务时，如果没有更好的本地性级别，暂时不分配任务而是等待一小会以期望获得更好的本地性级别的等待时间。
        spark.locality.wait:调度作业时，如果没有更好的本地性级别，退而求其次调度到次好的本地性级别的等待时间，默认是3秒。该时间是所有本地性级别的等待时间(process-local, node-local, rack-local 和any)的默认值。当然我们也可以分别配置各个本地性级别的等待时间。
        spark.locality.wait.process:PROCESS_LOCAL本地性级别的等待时间，即尝试访问缓存在executor进程的内存中数据的等待时间，默认值是spark.locality.wait的值。
        spark.locality.wait.node: NODE_LOCAL本地性级别的等待时间，默认值是spark.locality.wait的值。比如，我们可以将该值设置为0，以跳过NODE_LOCAL本地性级别，而直接尝试RACK_LOCAL。
        spark.locality.wait.rack: RACK_LOCAL本地性级别的等待时间，默认值是spark.locality.wait的值。
        一般来说这几个参数默认的配置不需要修改，但如果task执行时间长且本地性级别差的话（可以通过WebUi观察到），我们可以调高这些参数，使系统等待更长的时间以满足更好的本地性来运行作业。


欢迎关注笔者微信公众号“三角兽”，了解更多数学、算法、大数据干货文章。

好帖 顶一个

补充内容 (2021-3-31 18:10):
優惠券印刷