val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
val rdd: RDD[String] = sc.textFile("F:\\JavaProjects\\SparkCore\\input\\3.txt")
makeRDD
源码,发现调用了parallelize
方法,使用了从外面传进来的numSlices
,如果创建rdd的时候没有指定默认分区数目,那么默认值为defaultParallelism=taskScheduler.defaultParallelism
如图所示:image-20210414234931007
image-20210414235215258
taskScheduler.defaultParallelism
源码,发现它这是一个特质,还有具体实现的子类,那么思路有了,去查看哪些类实现了这个特质:image-20210414235347268
TaskSchedulerlmpl
,发现这依旧是一个特质的方法,看来还有一层特质需要去挖掘image-20210414235531039
defaultParallelism
的子类,发现如下信息:image-20210414235634949
LocalSchedulerBackend
,接着去看LocalSchedulerBackend
的具体方法实现,这里的conf应该是创建SparkContext时候新建的SparkConf类型的对象,如果设置了spark.default.parallelism这个属性的话,那么numSlices的值应该是这个属性的值,如果没有设置的话,值将是totalCores:image-20210415000041654
image-20210415000458891
image-20210415000707314
parallelize
方法开始:image-20210415002149984
ParallelCollectionRDD
对象,继续查看ParallelCollectionRDD
,在对象中有一个slice方法,这个方法即是切片逻辑:def slice[T: ClassTag](seq: Seq[T], numSlices: Int): Seq[Seq[T]] = {
if (numSlices < 1) {
throw new IllegalArgumentException("Positive number of partitions required")
}
def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {
(0 until numSlices).iterator.map { i =>
val start = ((i * length) / numSlices).toInt
val end = (((i + 1) * length) / numSlices).toInt
(start, end)
}
}
seq match {
case r: Range =>
positions(r.length, numSlices).zipWithIndex.map { case ((start, end), index) =>
// If the range is inclusive, use inclusive range for the last slice
if (r.isInclusive && index == numSlices - 1) {
new Range.Inclusive(r.start + start * r.step, r.end, r.step)
}
else {
new Range(r.start + start * r.step, r.start + end * r.step, r.step)
}
}.toSeq.asInstanceOf[Seq[Seq[T]]]
case nr: NumericRange[_] =>
// For ranges of Long, Double, BigInteger, etc
val slices = new ArrayBuffer[Seq[T]](numSlices)
var r = nr
for ((start, end) <- positions(nr.length, numSlices)) {
val sliceSize = end - start
slices += r.take(sliceSize).asInstanceOf[Seq[T]]
r = r.drop(sliceSize)
}
slices
case _ =>
val array = seq.toArray // To prevent O(n^2) operations for List etc
positions(array.length, numSlices).map { case (start, end) =>
array.slice(start, end).toSeq
}.toSeq
}
}
仔细分析以上方法,发现我们的List在模式匹配中匹配到了_情况,所以核心逻辑在内层函数positions
处封装:
def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {
(0 until numSlices).iterator.map { i =>
val start = ((i * length) / numSlices).toInt
val end = (((i + 1) * length) / numSlices).toInt
(start, end)
}
}
length为集合的长度,numSlices为分区数,每一个分区的计算逻辑如下:
即[start, end),前闭后开
举个简单例子,假设有一个集合List(1, 2, 3, 4),有3个分区,
那么分区和数据如下:
分区0:[0, 1) 1
分区1:[1, 2) 2
分区2:[2, 4) 3 4
当从集合中创建rdd时,分区数如果不设置,那么默认分区数将遵循以下规则:
默认分区切片逻辑如下:
textFile
源码,我们发现分区参数不再是numSplices,而是minPartitions,从字面意义上都代表着这个参数代表了最小分区数,如果我们不传值的话,他默认还是defaultMinPartitions
,接下来去看defaultMinPartitions
:image-20210415085646084
image-20210415091226838
defaultMinPartitions
中我们发现了熟悉的defalutParallelism
,这不就是使用集合创建RDD时候的默认分区数嘛,比较逻辑就是这个数和2取小的textFile
源码,发现最终是新建了hadoopFile
对象:image-20210415085646084
hadoopFile
源码,又发现底层新建了HadoopRDD
,找到了RDD的子类,感觉靠谱一点:image-20210415085743140
HadoopRDD
中我们找到所有RDD具有的特性,getPartitions方法:image-20210415085912983
仔细一看,这不是hadoop MapReduce的文件切片代码吗?到底是不是呢?让我们一探究竟。
image-20210415090115369
image-20210415090201559
得来全不费工夫,发现了熟悉的FileInputFormat,来让我们看一下具体的实现逻辑:
/** Splits files returned by {@link #listStatus(JobConf)} when
* they're too big.*/
public InputSplit[] getSplits(JobConf job, int numSplits)
throws IOException {
Stopwatch sw = new Stopwatch().start();
FileStatus[] files = listStatus(job);
// Save the number of input files for metrics/loadgen
job.setLong(NUM_INPUT_FILES, files.length);
long totalSize = 0; // compute total size
for (FileStatus file: files) { // check we have valid files
if (file.isDirectory()) {
throw new IOException("Not a file: "+ file.getPath());
}
totalSize += file.getLen();
}
long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
long minSize = Math.max(job.getLong(org.apache.hadoop.mapreduce.lib.input.
FileInputFormat.SPLIT_MINSIZE, 1), minSplitSize);
// generate splits
ArrayList<FileSplit> splits = new ArrayList<FileSplit>(numSplits);
NetworkTopology clusterMap = new NetworkTopology();
for (FileStatus file: files) {
Path path = file.getPath();
long length = file.getLen();
if (length != 0) {
FileSystem fs = path.getFileSystem(job);
BlockLocation[] blkLocations;
if (file instanceof LocatedFileStatus) {
blkLocations = ((LocatedFileStatus) file).getBlockLocations();
} else {
blkLocations = fs.getFileBlockLocations(file, 0, length);
}
if (isSplitable(fs, path)) {
long blockSize = file.getBlockSize();
long splitSize = computeSplitSize(goalSize, minSize, blockSize);
long bytesRemaining = length;
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,
length-bytesRemaining, splitSize, clusterMap);
splits.add(makeSplit(path, length-bytesRemaining, splitSize,
splitHosts[0], splitHosts[1]));
bytesRemaining -= splitSize;
}
if (bytesRemaining != 0) {
String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations, length
- bytesRemaining, bytesRemaining, clusterMap);
splits.add(makeSplit(path, length - bytesRemaining, bytesRemaining,
splitHosts[0], splitHosts[1]));
}
} else {
String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,0,length,clusterMap);
splits.add(makeSplit(path, 0, length, splitHosts[0], splitHosts[1]));
}
} else {
//Create empty hosts array for zero length files
splits.add(makeSplit(path, 0, length, new String[0]));
}
}
sw.stop();
if (LOG.isDebugEnabled()) {
LOG.debug("Total # of splits generated by getSplits: " + splits.size()
+ ", TimeTaken: " + sw.elapsedMillis());
}
return splits.toArray(new FileSplit[splits.size()]);
}
protected long computeSplitSize(long goalSize, long minSize,
long blockSize) {
return Math.max(minSize, Math.min(goalSize, blockSize));
}
首先有这么几个变量:
核心就是splitSize的计算,计算逻辑 = Math.max(minSize, Math.min(goalSize, blockSize),blockSize就是块大小,hadoop3.x都是128M,在我们计算出这个每个切片需要读取的字节数大小之后,就开始进入了切片逻辑:
long bytesRemaining = length;
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,
length-bytesRemaining, splitSize, clusterMap);
splits.add(makeSplit(path, length-bytesRemaining, splitSize,
splitHosts[0], splitHosts[1]));
bytesRemaining -= splitSize;
}
if (bytesRemaining != 0) {
String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations, length
- bytesRemaining, bytesRemaining, clusterMap);
splits.add(makeSplit(path, length - bytesRemaining, bytesRemaining,
splitHosts[0], splitHosts[1]));
}
bytesRemaining就是我们整个文件的剩余字节数多少,初始状态下等于文件字节数,使用while循环除以splitSize的大小如果大于SPLIT_SLOP=1.1的话,可以切片,切片之后bytesRemaining进行自减:
切片索引为(0, splitSize),(splitSize, splitSize * 2),依次类推,最后切出来几个片,就是最终的分区数,此时需要注意,切片读取的时候是按照行读取的,所以有些分区的数据会发生丢失,举个例子:
0 1 2 3 4 5 6 7
a b c d e f x x
8 9 10 11
j h x x
12 13 14 15 16
i j k x x
17 18 19
p i u
x x 代表\r \n
原始数据如下:
abcdef
jh
ijk
piu
假设传进去minPartitions = 3,按照上面的逻辑进行计算splitSize = 6,那么一个切片应该放的字节数 = 6:
a b c d e f
j h i j k
,从第6个字节读到12个字节,由于是按行读取,13 - 16位的字节被第二片已经读掉了,导致第三片只是读到了p i u
p i u
用程序打个断点看我们分析的是否正确:
image-20210415100354216
image-20210415100413100
image-20210415100431927
image-20210415100618407
本文分享自 Tyrant Lucifer 微信公众号,前往查看
如有侵权,请联系 cloudcommunity@tencent.com 删除。
本文参与 腾讯云自媒体同步曝光计划 ,欢迎热爱写作的你一起参与!