SparkContext和RDD的说明

一.Spark上下文

  1.做用:链接Spark集群,用户建立RDD、累加器和广播。shell

  2.RDD:Resilient Distributed Dataset,弹性式分布式数据集,有4种类型,以下:
    a.建立RDD:3类(parallelize:将Seq序列数据转化为RDD、textFile将外部文件转化RDD、makeRDD将Seq序列数据转化为RDD并可指定分区)
    b.转换操做:将RDD转换为另外一个RDD,转换操做不触发运算
    c.行动操做:触发运算,执行Job
    d.控制操做:对RDD的cache(缓存应用)apache

二.RDD的说明

  1.RDD表明一个不可变的,可并行操做的元素分区集合。缓存

  2.内部表现形式上,RDD可分为如下5部分:
    a.partition分区:获取partition分区列表,实际上就是切分(inputSplit)对象
    b.partitioner分区函数:针对每一个Split提供相对应的分区函数(partitioner)
    c.dependencies依赖关系:描述RDD之间的依赖关系,获取RDD的依赖关系的列表
    d.hash分区定义:针对KV类型的RDD,默认采起Hashpartitioon
    e.preferredlocations首选位置:用于计算每一个split的首选位置列表(例如用于HDFS文件块的位置)分布式

三.rdd建立partition说明

  1.local版:spark-shell --Master local[n]
    a.【parallelize】-->默认建立的分区数为"n"
    b.【textFile】 -->加载本地文件:默认建立的分区数为最大为2,若是n=1<2,则为1
          -->hdfs文件系统:加载的分区数为block个数;注:block块的大小应大于10%(inputsplit知识点);更改分区数必须大于等于block数
  2.spark集群 spark-shell --Master spark://master:7077
    a.【parallelize】-->默认建立的分区数为Cores的总数==Executor数
    b.【textFile】 -->加载本地文件:默认建立的分区数为最大为2,若是n=1<2,则为1
          -->hdfs文件系统:加载的分区数为block个数;注:block块的大小应大于10%(inputsplit知识点);更改分区数必须大于等于block数
  3.yarn集群 spark-shell --Master yarn
    a.【parallelize】-->默认建立的分区数为yarn集群的Executor数
    b.【textFile】 -->加载本地文件:默认建立的分区数为最大为2,若是n=1<2,则为1
          -->hdfs文件系统:加载的分区数为block个数;注:block块的大小应大于10%(inputsplit知识点);更改分区数必须大于等于block数  函数

四.切分大小源码分析

val rdd = sc.textFile("file:///D:/测试数据/spark_wordcount.txt")
-->【SparkContext.scala】def textFile(path: String,minPartitions: Int = defaultMinPartitions): RDD[String]
  说明:def defaultMinPartitions: Int = math.min(defaultParallelism, 2)
-->调用hadoopFile() 方法
  -->new HadoopRDD()
    -->经过getPartitions()方法获取partitions分区数
      -->调用【org.apache.hadoop.mapred.FileInputFormat】Hadoop老版本的API的getSplits()
        -->计算规则以下:
          1.首先计算goalSize(目标大小)
            long goalSize = totalSize / (long)(numSplits == 0 ? 1 : numSplits);
          2.获取SplitSize(切分大小)
            long splitSize = this.computeSplitSize(goalSize, minSize, blockSize);
            computeSplitSize --> Math.max(minSize, Math.min(goalSize, blockSize));oop

五.preferredlocations首选位置

获取RDD首选位置,首先经过RDD依赖获得HadoopRDD,调用preferredLocations方法
  scala> val rdd = sc.textFile("/worldcount/test1.txt")
    rdd: org.apache.spark.rdd.RDD[String] = /worldcount/test1.txt MapPartitionsRDD[3] at textFile at <console>:24源码分析

  scala> rdd.dependencies
    res17: Seq[org.apache.spark.Dependency[_]] = List(org.apache.spark.OneToOneDependency@bd9abda)测试

  scala> rdd.dependencies(0)
    res18: org.apache.spark.Dependency[_] = org.apache.spark.OneToOneDependency@bd9abdathis

  scala> val rdd1 = rdd.dependencies(0).rdd
    rdd1: org.apache.spark.rdd.RDD[_] = /worldcount/test1.txt HadoopRDD[2] at textFile at <console>:24spa

  scala> rdd1.preferredLocations(rdd.partitions(0))
    res19: Seq[String] = ArraySeq(master, slave1)

  scala> rdd1.preferredLocations(rdd.partitions(1))
    res20: Seq[String] = ArraySeq(master, slave1)