Spark并行度的设定

今天有同事问起Spark中spark.default.parallelism参数的意义，以及该如何设置。故在这里留个记录，算是做个小结。

Spark并行度设置的相关因素

Spark并行度的设置在Spark任务中是常常会谈及的问题，它是由partition的数量决定的。而partition的数量是由不同的因素决定的，它和资源的总cores、spark.default.parallelism参数、读取数据源的类型等有关系，在不同情况下，某个因素就起着主要的作用。下面看下Spark读取HDFS文本的并行度设置。

Spark读取HDFS文本的并行度设置

spark读取HDFS文本确定parition的方式，和前辈mapreduce的方式核心原理是一致的，只是在获取defaultMinPartitions的时候有所不同。

  def textFile(path: String,minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {assertNotStopped()hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],minPartitions).map(pair => pair._2.toString).setName(path)}

如果spark.default.parallslism有设置，defaultPartitions就会取设置的这个值。如果没有设置，则会根据分配给任务的总的cores数量和2比较后取最大值：

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

defaultMinPartitions会再取个最小值：

def defaultMinPartitions: Int = math.min(defaultParallelism, 2)

hadoopFile()中实例化了HadoopRDD，计算parition数量会调用getPartitions()，然后熟悉的过程就发生了。

  override def getPartitions: Array[Partition] = {val jobConf = getJobConf()// add the credentials here as this can be called before SparkContext initializedSparkHadoopUtil.get.addCredentials(jobConf)val inputFormat = getInputFormat(jobConf)val inputSplits = inputFormat.getSplits(jobConf, minPartitions)val array = new Array[Partition](inputSplits.size)for (i <- 0 until inputSplits.size) {array(i) = new HadoopPartition(id, i, inputSplits(i))}array}

HDFS文本属于FileInputFormat类型，所以这里就会动态调用FileInputFormat类的getSplits()方法，在这里算得了partition的数量。

先得到goalSize和minSize，供后面比较使用：

    long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);long minSize = Math.max(job.getLong(org.apache.hadoop.mapreduce.lib.input.FileInputFormat.SPLIT_MINSIZE, 1), minSplitSize);

文本是可切割的，那么喜蛋终于来了：

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]));}
}

来看computeSplitSize()方法，这里确定了一个partition的大小（blockSize是HDFS文件块的大小，默认128M）：

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

默认minSize很小，设为1，这里也假设是默认值不变。

如果goalSize > blockSize，则splitSize值取blockSize，肯定会有2个以上的partition。

如果goalSize < blockSize，则splitSize值取goalSize，会产生1-2个partition。

Spark读取HDFS文本的parition数量得到以后，并行度也就确定了。

如何设置Spark并行度才是合理的？

Spark并行度对于提高Spark任务的运行效率是非常关键的。合理设置Spark并行度可以从几个方面考虑：

1、充分利用任务资源（并行度略高于分配给任务的cpu资源数Executors * 每个Executor使用的cores）

2、平均每个parition的大小不要太大不要过小，一般在百兆较合适

3、根据实际机器的分配给任务的资源和任务需要计算的数据量大小，根据上面两点进行权衡设置。