当前位置：首页 > news >正文

从源码的角度告诉你 spark是怎样完成对文件切片

news 2025/7/17 21:13:38

1.说明

2.怎样设置默认切片数

2.1 RDD默认切片设置

2.2 SparkSQL默认切片设置

3. makeRDD 切片原理

4. textFile 切片原理

4.1 切片规则

4.2 怎样设置切片大小

4.3 测试代码

5.hadoopFile 切片原理

5.1 说明

5.2 切片规则

5.3 怎样设置切片大小

5.4 代码测试

5.5 minPartitions 在 CombineTextInputFormat 中的作用？

5.6 重点关注

1.说明

在spark中为我们提供了用来读取数据的方法
比如 makeRDD、parallelize、textFile、hadoopFile等方法

这些方法按照数据源可以分为两类 文件系统、Driver内存中的集合数据
当我们使用指定的方法读取数据后，会按照指定的切片个数对文件进行切片

2.怎样设置默认切片数

在我们在使用RDD的算子时，经常会遇到可以显式的指定切片个数，或者隐式的使用默认切片个数，下面会告诉我们，怎样设置默认切片个数

2.1 RDD默认切片设置

1.驱动程序中设置
val sparkconf: SparkConf = new SparkConf().setAppName("测试默认切片数").set("spark.default.parallelism","1000").setMaster("local[100]")2.spark-shell或spark-submit 设置
spark-shell \
--master yarn \
--name "spark-shell-tmp" \
--conf spark.default.parallelism=1000 \
--driver-memory 40G \
--executor-memory 40G \
--num-executors 40 \
--executor-cores 6 \3.不指定 spark.default.parallelism 参数时，将使用默认值local模式：local[100] :  100local      :  客户端机器核数集群模式(yarn)：2 或者 核数总和

源码：

查看默认切片数：

// 获取默认切片数
val parallelism = sc.defaultParallelism

2.2 SparkSQL默认切片设置

-- 设置默认切片数
set spark.sql.shuffle.partitions=1000;默认值：当不设置时，默认为200注意：spark.default.parallelism 只有在处理RDD时才会起作用，对SparkSQL的无效spark.sql.shuffle.partitions 则是对sparks SQL专用的设置

3. makeRDD 切片原理

可用通过 makeRDD算子将Driver中序列集合中数据转换成RDD，在转换的过程中，会根据指定的切片个数 和 集合索引对集合切片

切片规则：

根据集合长度和切片数将集合切分成若干子集合(和集合元素内容无关)

示例代码：

  test("makeRDD - 切片逻辑") {// 初始化 spark配置实例val sparkconf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("")// 初始化 spark环境对象val sc: SparkContext = new SparkContext(sparkconf)val rdd: RDD[(String, String)] = sc.makeRDD(List(("张飞1", "张飞java scala spark"), ("张飞2", "张飞java scala spark"), ("刘备3", "刘备java spark"), ("刘备4", "刘备java scala spark"), ("刘备5", "刘备scala spark"), ("关羽6", "关羽java scala spark"), ("关羽7", "关羽java scala"), ("关羽8", "关羽java scala spark"), ("关羽9", "关羽java spark")))// 查看每个分区的内容rdd.mapPartitionsWithIndex((i, iter) => {println(s"分区编号$i :${iter.mkString(" ")}");iter}).collect()rdd.getNumPartitionssc.stop()}

结果：

源码阅读：

1. 通过SparkContext创建rdd
def parallelize[T: ClassTag](
seq: Seq[T],
numSlices: Int = defaultParallelism): RDD[T] = withScope {
assertNotStopped()
new ParallelCollectionRDD[T](this, seq, numSlices, Map[Int, Seq[String]]())
}2. ParallelCollectionRDD类中的 getPartitions方法
override def getPartitions: Array[Partition] = {val slices = ParallelCollectionRDD.slice(data, numSlices).toArrayslices.indices.map(i => new ParallelCollectionPartition(id, i, slices(i))).toArray
}3. 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")}// TODO 通过 集合长度和切片数 获取每个切片的位置信息// 从这可以得出 对集合的切片只和 集合索引和切片数相关，和集合内容无关// 将 集合索引按照切片数 切分成若干元素def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {(0 until numSlices).iterator.map { i =>val start = ((i * length) / numSlices).toIntval 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 sliceif (r.isInclusive && index == numSlices - 1) {new Range.Inclusive(r.start + start * r.step, r.end, r.step)} else {new Range.Inclusive(r.start + start * r.step, r.start + (end - 1) * r.step, r.step)}}.toSeq.asInstanceOf[Seq[Seq[T]]]case nr: NumericRange[T] =>// For ranges of Long, Double, BigInteger, etcval slices = new ArrayBuffer[Seq[T]](numSlices)var r = nrfor ((start, end) <- positions(nr.length, numSlices)) {val sliceSize = end - startslices += r.take(sliceSize).asInstanceOf[Seq[T]]r = r.drop(sliceSize)}slices.toSeqcase _ =>val array = seq.toArray // To prevent O(n^2) operations for List etcpositions(array.length, numSlices).map { case (start, end) =>array.slice(start, end).toSeq}.toSeq}
}

4. textFile 切片原理

textFile使用的MapReduce框架中TextInputFormat类完成对文件切片和读取切片中数据