SPARK[RDD之partitions]

2023年11月25日 283次阅读 来源: 北风第一支

RDD是容错、并行的数据结构，具备分区的属性，这个分区可以是单机上分区也可以是多机上的分区，对于RDD分区的数量涉及到这个RDD进行并发计算的粒度。每一个分区都会在一个单独的task中执行。

可以为其指定分区个数，如果从hdfs文件创建的RDD，分区数和block数一致，如果从本地文件中创建RDD，默认是机器的cpu个数。

//不设置分区数量
>val rdd = sc.textFile("/home/reducer2/cluster/hadoop/readme.md")
>rdd.partitions.size
res0: INT = 8  //为cpu的个数

//设置分区个数
>val rdd = sc.textFile("/home/reducer2/cluster/hadoop/readme.md",6);

RDD的首选位置 preferredLocations
spark在执行任务时，会尽可能的把算子分配到离数据最近的节点上，减少数据的网络IO，当RDD生成的位置就是首选位置，如果是HDFS生成的RDD，那首选位置就是block所在在的节点。如果是经过转换后的RDD，则算子应该分配到RDD所在的节点上。

如果一个大文件40G，生成一个RDD,产生与block个数一样多的分区，默认一个block为128M，则会产生320个block（partitions）,假如共有5台natanode平均分配了这320个block,每个datanode上至少有64个分区。现在要对分区1进行计算，第一步则需要知道RDD分区1的首选位置:

val location = rdd.preferredLocations(rdd.dependecies(0))
res1:Set[String] =(192.168.110.1,192.168.110.2,192.168.110.3) //和block的复制数一致

而在真实的开发过程中，我们并没有去关注单个分区的执行的情况，而是类似于：

val rdds = sc.textFile("/.../..");
varl count = rdds.filter(line=>line.contains("error"))

在执行filter的时候，实际上就是每个分区都会执行这个函数，最后会生成一个新的RDD，新RDD的分区情况可能和先前的分区大致相同，在相同机器节点上，减少机器间的数据重排。

    原文作者：北风第一支
    原文地址: https://www.jianshu.com/p/e43fa6ee2e15
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞