本文共 4275 字,大约阅读时间需要 14 分钟。
Spark RDD主要由Dependency、Partition、Partitioner组成,Partition是其中之一。一份待处理的原始数据会被按照相应的逻辑(例如jdbc和hdfs的split逻辑)切分成n份,每份数据对应到RDD中的一个Partition,Partition的数量决定了task的数量,影响着程序的并行度,所以理解Partition是了解spark背后运行原理的第一步。
查看spark源码,trait Partition的定义很简单,序列号index和hashCode方法。Partition和RDD是伴生的,即每一种RDD都有其对应的Partition实现,所以,分析Partition主要是分析其子类。我们关注两个常用的子类,JdbcPartition和HadoopPartition。此外,RDD源码中有5个方法,代表其组成,如下: 第二个方法,getPartitions是数据源如何被切分的逻辑,返回值正是Partition,第一个方法compute是消费切割后的Partition的方法,所以学习Partition,要结合getPartitions和compute方法。
JdbcPartition例子
下面是Spark JdbcRDDSuite中一个例子val sc = new SparkContext("local[1]", "test") val rdd = new JdbcRDD( sc, () => { DriverManager.getConnection("jdbc:derby:target/JdbcRDDSuiteDb") }, // DATA类型为INTEGER "SELECT DATA FROM FOO WHERE ? <= ID AND ID <= ?", 1, 100, 3, (r: ResultSet) => { r.getInt(1) } ).count()
查看JdbcPartition实现,相比Partition,主要多了lower和upper这两个字段。 查看JdbcRDD的getPartitions,按照如上图所示算法将1到100分为3份(partition数量),结果为(1,33)、(34,66)、(67,100),封装为JdbcPartition并返回,这样数据切分的部分就完成了。 查看JdbcRDD的compute方法,逻辑清晰,将Partition强转为JdbcPartition,获取连接并预处理sql,将 例子中的”SELECT DATA FROM FOO WHERE ? <= ID AND ID <= ?”问号分别用Partition的lower和upper替换(即getPartitions切分好的(1,33)、(34,66)、(67,100))并执行查询。至此,JdbcPartition如何发挥作用就分析完了。HadoopPartition例子
举个简单例子val sc = new SparkContext("local[1]", "test")sc.textFile("hdfs://your-file-path").count()
Partition数量可以在初始化RDD时指定(如JdbcPartition例子),不指定的话(如HadoopPartition例子),则
读取配置,不同类型资源管理器取值不同,如下 了解了默认的partition数量,再看一些具体API的partition行为Spark API | partition数量 |
sc.parallelize(…) | sc.defaultParallelism |
sc.textFile(…) | max(传参, block数) |
val hbaseRDD = sc.newAPIHadoopRDD(…) | max(传参, block数) |
val jdbcRDD = new JdbcRDD(…) | 传参 |
filter(),map(),flatMap(),distinct() | 和父RDD相同 |
rdd.union(otherRDD) | rdd.partitions.size + otherRDD. partitions.size |
rdd.intersection(otherRDD) | max(rdd.partitions.size, otherRDD. partitions.size) |
rdd.subtract(otherRDD) | rdd.partitions.size |
rdd.cartesian(otherRDD) | rdd.partitions.size * otherRDD. partitions.size |
reduceByKey(),foldByKey(),combineByKey(), groupByKey() | 和父RDD相同 |
sortByKey() | 同上 |
mapValues(),flatMapValues() | 同上 |
cogroup(), join(), ,leftOuterJoin(), rightOuterJoin() | 所有父RDD按照其partition数降序排列,从partition数最大的RDD开始查找是否存在partitioner,存在则partition数由此partitioner确定,否则,所有RDD不存在partitioner,由spark.default.parallelism确定,若还没设置,最后partition数为所有RDD中partition数的最大值 |
上面的Partition行为我们从中挑一个细分析,就是sc.textFile(…, numPartitions)读取hdfs时的Partition数,上表给出的答案是numPartitions和block数较大者,如果不指定numPartitions,则numPartitions<=2, 分析这个问题,其实跟spark无关,要查看Hadoop源码FileInputFormat类中getSplits方法
指定numPartitions
totalSize为待处理文件总大小,numSplits就是我们所指定的numPartitions,得到了平均的文件大小goalSize,接下来 比较计算得到的goalSize和block大小blockSize,取其中较小者,再和minSize(由属性确定,默认值为0,则minSize默认值为1)取较大的。 假设待处理文件大小fSize=512M(视为一个大文件,不考虑1.1系数),block大小bSize=128M,sc.textFile(…, 3)
根据上面的公式goalSize=512M/3 > bSize=128M 取其较小者bSize,则按照bSize切分,split数=512M/128=4,即partition数=4sc.textFile(…, 5)
根据上面的公式goalSize=512M/5 < bSize=128M 取其较小者goalSize,则按照goalSize切分,split数=512M/(512M/5)=5,即partition数=5可见指定numPartitions,小于block数时无效,大于则生效。
不指定numPartitions
默认,传给FileInputFormat类getSplits方法的numSplits值是sc.defaultParallelism和2的较小值,所以spark.default.parallelism几乎是没用的,Partition数就是block数。那么为什么是这样的呢,感兴趣的同学看下这个上面分析了决定Partition数量的因数,接下来就该考虑Partition数量的影响以及合适的值。
Partition数量的影响
那么多少的partition数是合适的呢,这里我们参考spark doc给出的建议,。
Partition对应的是不同数据源的split逻辑,首先以JdbcPartition和HadoopPartition为例,介绍了Partition的组成,以及如何发挥作用,接下来分析了常见API的Partition行为,最后简单介绍了Partition数量的影响及调整。
参考:
注:图片中代码均为Spark、Hadoop源码,我稍作处理,如去掉log、metric等,使逻辑更清晰。