Spark学习总结一_spark引擎技术与应用心得

一、Spark运行架构

Spark框架核心是一个计算引擎，采用了标准的master-slave标准。其中Driver表示Master，负责管理整个集群的作业任务调度。Excutor是slave，负责实际执行任务。

二、任务提交流程

2.1 通用提交流程

2.2 Standalone模式

2.3 YARN模式

Spark应用程序提交到Yarn环境中执行的时候，一般会有两种部署执行的方式：Client和Cluster。两种模式，主要区别在于：Driver程序的运行节点。

2.4核心概念解释

三、程序解析

3.1 WordCount程序解析

3.1.1wordcount程序

object WordCount {

  def main(args: Array[String]): Unit = {

    val conf=new SparkConf().setMaster("local").setAppName("My app")

    //手动创建应用的上下文
    val sparkContext = new SparkContext(conf)

    // 进行wc的编程
    // 创建RDD  RDD[String]: RDD[一行内容]  HadoopRDD
    val rdd: RDD[String] = sparkContext.textFile("input")


    //  RDD[String]:  RDD[单词]
    val rdd1: RDD[String] = rdd.flatMap(line => line.split(" "))

    // RDD[(String, Int)]: RDD[(单词，1)]
    val rdd2: RDD[(String, Int)] = rdd1.map(word => (word, 1))


    // reduceByKey:  将相同key对应的values进行reduce运算！
    val result: RDD[(String, Int)] = rdd2.reduceByKey((value1, value2) => value1 + value2)

    // 对结果进行输出
   // println(result.collect().mkString(","))
   // println(result.collect().mkString(","))

    result.saveAsTextFile("output")

    // 运算完成后，需要关闭sparkContext
    sparkContext.stop()

  }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

3.1.2简写版

object WordCount2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("My app")
    //手动创建应用的上下文
    val sparkContext = new SparkContext(conf)
    //进行wc的编程
    //创建RDD RDD[String]:RDD[一行内容]
    val result:RDD[(String,Int)]=sparkContext.textFile("sparkday01/input")
      .flatMap(_.split(" "))
      .map((_,1))
      .reduceByKey(_+_)
    //对结果进行打印输出
    println(result.collect().mkString(","))
    //关闭sparkContext
    sparkContext.stop()
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

3.1.3 数据流图解

3.2 RDD的创建

3.2.1.从集合（内存）中创建RDD（parallelize和makeRDD方法都可以）

val sparkConf =new SparkConf().setMaster("local[*]").setAppName("spark")
val sparkContext = new SparkContext(sparkConf)
def testMakeRDD()={

      val list = List(1,2,3,4,5,6)
      val rdd1 = sparkContext.makeRDD(list,5)
    val rdd2 = sparkContext.parallelize(List(1,2,3,4)）
    }
1
2
3
4
5
6
7
8

从底层代码来讲，makeRDD方法就是parallize（）方法

def makeRDD[T: ClassTag](
    seq: Seq[T],
    numSlices: Int = defaultParallelism): RDD[T] = withScope {
  parallelize(seq, numSlices)
}
1
2
3
4
5

3.2.2.从外部存储文件创建RDD（textFile方法）

本地文件系统，所有Hadoop支持的数据集，比如HDFS和HBase等

val sparkConf =new SparkConf().setMaster("local[*]").setAppName("spark")
val sparkContext = new SparkContext(sparkConf)
//使用textFile方法从指定路径获取
val fileRDD: RDD[String] = sparkContext.textFile("input")
fileRDD.collect().foreach(println)
sparkContext.stop()
1
2
3
4
5
6

3.2.3.从其他RDD创建（转换算子）

3.4textFile分区策略源码解析

3.4.1 分区数源码追寻

 def textFile(
      path: String,
     //读取文件时，传递的参数为最小分区数，但不一定就是这个分区数，取决于hadoop读取文件时的分片规则
      minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {
    assertNotStopped()
    hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
      minPartitions).map(pair => pair._2.toString).setName(path)
  }
1
2
3
4
5
6
7
8

//比较defaultParralelism的和2的最小值，默认的SparkConf中没有设置spark.default.parallelism，默认defaultParallelism=totalCores(当前本地集群可以用的总核数)，目的为了最大限度并行运行，取默认值和2的最小值，defaultMinPartitions和minPartitions不是最终分区数，但是会影响最终分区数！最终分区数，还取决于切片数！
def defaultMinPartitions: Int = math.min(defaultParallelism, 2)
1
2

3.4.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 {
        // 调用输入格式（org.apache.hadoop.mapred.TextInputFormat）进行切片，切片时，   minPartitions=2
      val allInputSplits = getInputFormat(jobConf).getSplits(jobConf, minPartitions)
      
        // 是否过滤空切片后的切片集合
        val inputSplits = if (ignoreEmptySplits) {
        allInputSplits.filter(_.getLength > 0)
      } else {
        allInputSplits
      }
        
        // 如果切的是1片，且是针对文件的切片，做特殊处理
      if (inputSplits.length == 1 && inputSplits(0).isInstanceOf[FileSplit]) {
        val fileSplit = inputSplits(0).asInstanceOf[FileSplit]
        val path = fileSplit.getPath
        if (fileSplit.getLength > conf.get(IO_WARNING_LARGEFILETHRESHOLD)) {
          val codecFactory = new CompressionCodecFactory(jobConf)
          if (Utils.isFileSplittable(path, codecFactory)) {
            logWarning(s"Loading one large file ${path.toString} with only one partition, " +
              s"we can increase partition numbers for improving performance.")
          } else {
            logWarning(s"Loading one large unsplittable file ${path.toString} with only one " +
              s"partition, because the file is compressed by unsplittable compression codec.")
          }
        }
      }
        
        // 分区数=过滤后的切片数
      val array = new Array[Partition](inputSplits.size)
      for (i <- 0 until inputSplits.size) {
        array(i) = new HadoopPartition(id, i, inputSplits(i))
      }
      array
    } catch {
      case e: InvalidInputException if ignoreMissingFiles =>
        logWarning(s"${jobConf.get(FileInputFormat.INPUT_DIR)} doesn't exist and no" +
            s" partitions returned from this path.", e)
        Array.empty[Partition]
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

切片机制源码

由上述源码可知分区数=过滤后的切片数，因此查看切片方法源码

public InputSplit[] getSplits(JobConf job, int numSplits)
    throws IOException {
  
        
        // 当前切片的数据的总大小
    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();
    }

        // 计算 goalsize(期望每片大小)，numSplits受并行度影响，如果设置了则按照设置个数来算，没设置就按照并行度和2取最小值def defaultMinPartitions: Int = math.min(defaultParallelism, 2)
    long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
        
        // 默认为1，调节 org.apache.hadoop.mapreduce.lib.input. FileInputFormat.SPLIT_MINSIZE， 改变minSize
     
    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)) {
            // 获取文件的块大小，块大小在上传文件时，指定，如果不指定，默认 128M
          long blockSize = file.getBlockSize();
            
            // 计算片大小   一般等于  blockSize
          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;
          }

            // 剩余部分 <=片大小1.1倍，整体作为1片
          if (bytesRemaining != 0) {
            String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations, length
                - bytesRemaining, bytesRemaining, clusterMap);
            splits.add(makeSplit(path, length - bytesRemaining, bytesRemaining,
                splitHosts[0], splitHosts[1]));
          }
        } else {
            //如果文件不可切，整个文件作为1片
          String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,0,length,clusterMap);
          splits.add(makeSplit(path, 0, length, splitHosts[0], splitHosts[1]));
        }
      } else { 
          // 文件长度为0，创建一个空切片
        //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.now(TimeUnit.MILLISECONDS));
    }
    return splits.toArray(new FileSplit[splits.size()]);
  }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

long splitSize = computeSplitSize(goalSize, minSize, blockSize);
// 在处理大数据时，goalsize一般很大，（例如10T/100）一般情况下，blockSize作为片大小
return Math.max(minSize, Math.min(goalSize, blockSize));
//下面是hadoop中FileInputFormat的切片大小公式，很类似
Math.max（minSize，Math.min（maxSize，blockSize))其中minsize默认值为1，maxsize非常大
1
2
3
4
5

总结：从分析源码中获取的关键信息为，分区数=切片数，切片大小计算方法为max(minSize, Math.min(goalSize, blockSize));查看源码得知minSize=1，blockSize默认为128M（本地模式32M）

3.4.3 textFile分区实测解析

 @Test
  def TestTextFile(): Unit ={
    val result:RDD[(String,Int)]=sparkContext.textFile("input1",6)
      .flatMap(_.split(" "))
      .map((_,1))
      .reduceByKey(_+_)
    //对结果进行打印输出
    //println(result.collect().mkString(","))
    result.saveAsTextFile("output")
  }
input1中有aa.txt=351b bb.txt=349b文件goalSize=700/6=116.7b
切片大小为max（1，min（333.3b，32m））=116.7b
切片数为351/116.7=3+余数 349/116.7=2+余数，剩余部分小于1.1倍的116.7，因此放在一个分区内，总共有0-5六个partition
1
2
3
4
5
6
7
8
9
10
11
12
13

3.5 makeRDD分区策略源码解析

3.5.1分区数源码分析

def makeRDD[T: ClassTag](
      seq: Seq[T],
    //numSlices: 控制集合创建几个分区！
      numSlices: Int = defaultParallelism): RDD[T] = withScope {
    parallelize(seq, numSlices)
  }
1
2
3
4
5
6

def defaultParallelism: Int = {
    assertNotStopped()
    taskScheduler.defaultParallelism
  }
1
2
3
4

//抽象方法
def defaultParallelism(): Int
1
2

ctl+H找到实现类

override def defaultParallelism(): Int = backend.defaultParallelism()
1

点击defaultParallelism()

/**
 * A backend interface for scheduling systems that allows plugging in different ones under
 * TaskSchedulerImpl. We assume a Mesos-like model where the application gets resource offers as
 * machines become available and can launch tasks on them.
 */
private[spark] trait SchedulerBackend {
  private val appId = "spark-application-" + System.currentTimeMillis

  def start(): Unit
  def stop(): Unit
  def reviveOffers(): Unit
  def defaultParallelism(): Int
1
2
3
4
5
6
7
8
9
10
11
12

ctl+h找到SchedulerBackend的实现类CoarseGrainedSchedulerBackend

// 默认defaultParallelism=totalCores(当前本地集群可以用的总核数)，目的为了最大限度并行运行！
override def defaultParallelism(): Int = {
    //点击getInt方法查看
    conf.getInt("spark.default.parallelism", math.max(totalCoreCount.get(), 2))
  }
1
2
3
4
5

def getInt(key: String, defaultValue: Int): Int = catchIllegalValue(key) {
    getOption(key).map(_.toInt).getOrElse(defaultValue)
  }
1
2
3

默认defaultParallelism=totalCores(当前本地集群可以用的总核数)，目的为了最大限度并行运行！standalone / YARN模式， totalCores是Job申请的总的核数！

3.5.2分区策略

def slice[T: ClassTag](seq: Seq[T], numSlices: Int): Seq[Seq[T]] = {
    //检查分区数是否合法
    if (numSlices < 1) {
      throw new IllegalArgumentException("Positive number of partitions required")
    }
    // Sequences need to be sliced at the same set of index positions for operations
    // like RDD.zip() to behave as expected
    
    /*length:6
      numSlices:4    		
    */
    def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {
        /*
          (0 until numSlices) =   [0,4)
          	0: (0,1)
          	1: (1,3)
          	2: (3,4)
          	3: (4,6)
        */
      (0 until numSlices).iterator.map { i =>
        val start = ((i * length) / numSlices).toInt
        val end = (((i + 1) * length) / numSlices).toInt
        (start, end)
      }
    }
    ...... 对range类型进行特殊处理
    
    // 非Ranger，按此方法处理
      case _ =>
       //Array（1，2，3，4,5,6）
        val array = seq.toArray // To prevent O(n^2) operations for List etc
    // 返回  ｛(0,1)，(1,3)，(3,4),(4,6)｝
    //         1,(2,3),4,(5,6)
        positions(array.length, numSlices)
    .map { case (start, end) =>
            array.slice(start, end).toSeq
        // 1,(2,3),4,(5,6)
        }.toSeq
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

3.5.3 makeRD创建实例分析

class RDDTest {
//本地集群总核数取决于local[N] local：1核local[2]: 2核 local[*] : 所有核
  val sparkContext = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("My app"))
//由于本机cpu为4核，makeRDD方法没有设置并行度，按照local[*]参数来配置分区，因此该例计算文件个数为4个
  @Test
  def testMakeRDD()={

      val list = List(1,2,3,4,5,6)
      val rdd1 = sparkContext.makeRDD(list)
    rdd1.saveAsTextFile("output")
    }
 }

partition-00000	1
partition-00001	2,3
partition-00002	4
partition-00003	5,6
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17