一.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)