SparkContext和RDD的说明

一.Spark上下文

　　1.作用：连接Spark集群，用户创建RDD、累加器和广播。

　　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(缓存应用)

二.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));

五.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@bd9abda

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

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

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