Spark RDD 分布式弹性数据集

abstract class RDD[T: ClassTag](
    @transient private var _sc: SparkContext,
    @transient private var deps: Seq[Dependency[_]]
  ) extends Serializable with Logging {

// rdd维护这sparkContext上下文环境和rdd的依赖关系
private def sc: SparkContext = {
    if (_sc == null) {
      throw new SparkException(...)
    }
    _sc
}

/** Construct an RDD with just a one-to-one dependency on one parent */
  def this(@transient oneParent: RDD[_]) =
    this(oneParent.context, List(new OneToOneDependency(oneParent)))

// 1. 实现compute函数实现rdd每个分区的迭代计算
// Implemented by subclasses to compute a given partition.
 def compute(split: Partition, context: TaskContext): Iterator[T]

/**
   * Implemented by subclasses to return the set of partitions in this RDD. This method will only
   * be called once, so it is safe to implement a time-consuming computation in it.
   *
   * The partitions in this array must satisfy the following property:
   *   `rdd.partitions.zipWithIndex.forall { case (partition, index) => partition.index == index }`
   */
  // 2. 每个rdd都有一个或多个分区
  protected def getPartitions: Array[Partition]

/**
   * Implemented by subclasses to return how this RDD depends on parent RDDs. This method will only
   * be called once, so it is safe to implement a time-consuming computation in it.
   */
  // 3. rdd维护这自己的依赖关系
  protected def getDependencies: Seq[Dependency[_]] = deps

  /**
   * Optionally overridden by subclasses to specify placement preferences.
   */
  // 4. rdd可以根据数据的位置优化分区的计算位置，提高效率
  protected def getPreferredLocations(split: Partition): Seq[String] = Nil

  /** Optionally overridden by subclasses to specify how they are partitioned. */
  // 5. rdd都有自己的分区器，表明他们分区的方式
  @transient val partitioner: Option[Partitioner] = None
}

def parallelize[T: ClassTag](
    seq: Seq[T],
    numSlices: Int = defaultParallelism): RDD[T] = withScope {
  assertNotStopped()
  new ParallelCollectionRDD[T](this, seq, numSlices, Map[Int, Seq[String]]())
}

TaskSchedulerImpl类：
override def defaultParallelism(): Int = backend.defaultParallelism()
CoarseGrainedSchedulerBackend类：
override def defaultParallelism(): Int = {
    conf.getInt("spark.default.parallelism", math.max(totalCoreCount.get(), 2))
}

override def getPartitions: Array[Partition] = {
    val jobConf = getJobConf()
    // add the credentials here as this can be called before SparkContext initialized
    SparkHadoopUtil.get.addCredentials(jobConf)
    try {
      // 从这里可以看出allInputSplits来自getSplits
      val allInputSplits = getInputFormat(jobConf).getSplits(jobConf, minPartitions)
      val inputSplits = if (ignoreEmptySplits) {
        allInputSplits.filter(_.getLength > 0)
      } else {
        allInputSplits
      }
      // 分区数为inputSplits.size
      val array = new Array[Partition](inputSplits.size)
      for (i <- 0 until inputSplits.size) {
        array(i) = new HadoopPartition(id, i, inputSplits(i))
      }
      array
    } catch {
       ...
    }
  }

FileInputFormat类：
public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
        ...
        long totalSize = 0L;
        FileStatus[] arr$ = files;
        int len$ = files.length;
        // totalSize 为所有文件的总字节数
        for(int i$ = 0; i$ < len$; ++i$) {
            FileStatus file = arr$[i$];
            if (file.isDirectory()) {
                throw new IOException("Not a file: " + file.getPath());
            }

            totalSize += file.getLen();
        }
        // 对totalSize切分后的字节数goalSize，numSplits默认为min(2,totalCores)=2
        long goalSize = totalSize / (long)(numSplits == 0 ? 1 : numSplits);
        // minSize为配置的mapreduce切分值
        long minSize = Math.max(job.getLong("mapreduce.input.fileinputformat.split.minsize", 1L), this.minSplitSize);
        ....
        for(int i$ = 0; i$ < len$; ++i$) {
        ....
            else {
                    long blockSize = file.getBlockSize();
                    long splitSize = this.computeSplitSize(goalSize, minSize, blockSize);

                    long bytesRemaining;
                    String[][] splitHosts;
                    // 如果在切片大小1.1倍内的都不会被切分
                    for(bytesRemaining = length; (double)bytesRemaining / (double)splitSize > 1.1D; bytesRemaining -= splitSize) {
                        splitHosts = this.getSplitHostsAndCachedHosts(blkLocations, length - bytesRemaining, splitSize, clusterMap);
                        splits.add(this.makeSplit(path, length - bytesRemaining, splitSize, splitHosts[0], splitHosts[1]));
                    }

                    if (bytesRemaining != 0L) {
                        splitHosts = this.getSplitHostsAndCachedHosts(blkLocations, length - bytesRemaining, bytesRemaining, clusterMap);
                        splits.add(this.makeSplit(path, length - bytesRemaining, bytesRemaining, splitHosts[0], splitHosts[1]));
                    }
                }
            }
        }

        sw.stop();
        if (LOG.isDebugEnabled()) {
            LOG.debug("Total # of splits generated by getSplits: " + splits.size() + ", TimeTaken: " + sw.elapsedMillis());
        }

        return (InputSplit[])splits.toArray(new FileSplit[splits.size()]);
    }

protected long computeSplitSize(long goalSize, long minSize, long blockSize) {
        return Math.max(minSize, Math.min(goalSize, blockSize));
    }

override def getPreferredLocations(split: Partition): Seq[String] = {
  val hsplit: InputSplit = split.asInstanceOf[HadoopPartition].inputSplit.value
  val locs = hsplit match {
    case lsplit: InputSplitWithLocationInfo =>
      HadoopRDD.convertSplitLocationInfo(lsplit.getLocationInfo)
    case _ => None
  }
  // 返回ip不为localhost的列表
  locs.getOrElse(hsplit.getLocations.filter(_ != "localhost"))
}

val n = math.min(100000L * slices, Int.MaxValue).toInt // avoid overflow
val listRDD = spark.sparkContext.parallelize(1 until n, slices)
val mapRDD = listRDD.map(i => (i, i * i))
val groupRDD = mapRDD.groupByKey()

def defaultPartitioner(rdd: RDD[_], others: RDD[_]*): Partitioner = {
 ...
  // 如果存在配置并发配置，将并发配置作为默认分区数，否则上个rdd的最大值
  val defaultNumPartitions = if (rdd.context.conf.contains("spark.default.parallelism")) {
    rdd.context.defaultParallelism
  } else {
    rdds.map(_.partitions.length).max
  }

  // If the existing max partitioner is an eligible one, or its partitions number is larger
  // than the default number of partitions, use the existing partitioner.
  if (hasMaxPartitioner.nonEmpty && (isEligiblePartitioner(hasMaxPartitioner.get, rdds) ||
      defaultNumPartitions < hasMaxPartitioner.get.getNumPartitions)) {
    hasMaxPartitioner.get.partitioner.get
  } else {
    // 默认返回HashPartitioner
    new HashPartitioner(defaultNumPartitions)
  }
}