应用可以通过使用Spark提供的库获得Spark集群的计算能力，这些库都是Scala编写的，但是Spark提供了面向各种语言的API，例如Scala、Python、Java等，所以可以使用以上语言进行Spark应用开发。 Spark的API主要由两个抽象部件组成：SparkContext和RDD，应用程序通过这两个部件和Spark进行交互，连接到Spark-集群并使用相关资源。
是定义在Spark库中的一个类，作为Spark库的入口。包含应用程序main（）方法的Driver program通过SparkContext对象访问Spark，因为SparkContext对象表示与Spark集群的一个连接。每个Spark应用都有且只有一个激活的SparkContext类实例，如若需要新的实例，必须先让当前实例失活。 （在shell中SparkContext已经自动创建好，就是sc）
实例化SparkContext：
val sc = new SparkContext()弹性分布式数据集（Resilient Distributed Dataset）
并行分布在整个集群中 把指定路径下的文本文件加载到lines这个RDD中，这个lines就是一个RDD，代表是就是整个文本文件
val lines = sc.textFile("home/haha/helloSpark.txt")即使这个文件超大，分片存储在多台机器上，操作时可以直接使用RDD对整体文件进行操作
RDD是Spark分发数据和计算的基础抽象类 例如：lines.count() 在.count()的函数操作是在RDD数据集上的，而不是对某一具体分片
一个RDD是一个不可改变的分布式集合对象 就lines来说，如果我们对其所代表的源文件进行了增删改操作，则相当于生成了一个新的RDD，来存放修改后的数据集
Spark中所有的计算都是通过RDD的创建、转换，操作完成的
一个RDD内部由许多partitions（分片）组成 partitions： 每个分片包括一部分数据，分片可在集群不同节点上计算 分片是Spark并行处理的单元，Spark顺序的，并行的处理分片
1.把一个存在的集合传给SparkContext的parallize()方法（一般测试的时候使用这个方法）
val rdd = sc.parallelize(Array(1,2,2,1),4) //参数1：待并行化处理的集合；参数2：分区个数 rdd.count() //查看个数 rdd.foreach(print) //查看RDD的所有对象 //注意：每次foreach出来数值顺序会不一样，因为数据存储于4个分区，从哪个分区读取数值是随机的2.加载外部数据集
//用textFile方法加载 //该方法返回一个RDD，该RDD代表的数据集每个元素都是一个字符串，每个字符串代表输入文件中的一行 val rddText = sc.textFile("helloSpark.txt") //用wholeTextfiles方法加载 //这个方法读取目录下的所有文本文件，然后返回一个KeyValue对RDD（每一个键值对对应一个文件，key为文件路径，value为文件内容） val rddW = sc.wholeTextFile("path/to/my-data/*.txt") //用sequenceFile方法加载 //此方法要求从SequenceFile文件中获取键值对数据，返回一个KeyValue对RDD（使用此方法时，还需要提供类型） val rdd = sc.sequenceFile[String,String]("some-file")Transformation（转换）—– 从之前的RDD中构建一个新的RDD
逐元素 map()：接收函数后，把函数应用到RDD的每一个元素，返回新的RDD
val lines1 = sc.parallelize(Array("hello","spark"，"hello","world","!")) val lines2 = lines1.map(word=>(word,1)) lines2.foreach(println) //打印出： (hello,1) (spark,1) (hello,1) (world,1) (!,1)filter()：接收函数后，返回只包含满足filter()函数的元素的新RDD
val lines3 = lines.filter(word=>word.contains("hello")) lines3.foreach(println) //打印出： hello helloflatMap()：对每个输入元素，输出多个输出元素，也就是将RDD中元素flat（压扁）后返回一个新的RDD
val input = sc.textFile('/home/haha/helloSpark.txt') val lines4 = input.flatMap(line=>line.split(" ")) //用空格切分一行的单词集合运算 RDD支持数学集合的计算，例如并集、交集计算
val rdd1 = sc.parallelize(Array("coffe","coffe","panda","monkey","tea")) val rdd2 = sc.parallelize(Array("coffe","monkey","kitty")) val rdd_distinct = rdd1.distinct(rdd2) //合并，去重 val rdd_union = rdd1.union(rdd2) // 并集 val rdd_inter = rdd1.intersection（rdd2） // 交集 val rdd_sub = rdd1.subtract(rdd2) // rdd1中有rdd2中没有的Action（行动） —– 在RDD上计算出一个结果，并且把结果返回给driver program或保存在文件系统。例如：count()，save
RDD几个常用的Action：
举例说说几个重要操作：
reduce()：接收一个函数后，作用在RDD两个类型相同的元素上，返回新元素；可以实现RDD中元素的累加、计数，和其它类型的聚集操作。
val rdd = sc.parallelize(Array(1,2,3,3)) rdd.collect() //数组形式输出 //打印出： Array[Int] = Array(1,2,3,3) rdd.reduce((x,y)=>x+y) //以x，y代表同种类型数据 //打印出： Int = 9 //因为该RDD表示的数组中int类型数据总和为9collect()：遍历整个RDD，向driver program返回RDD的内容，需要单机内存能够容纳下（因为数据要拷贝给dirver，测试用该函数，方便看到数据所有内容）。在处理大量数据时，使用saveAsTextFile()等
take(n)：返回RDD的n个元素（同时尝试访问最少的partitions），返回结果是无序的（测试时使用） —– 测试时，随机取n个元素
top(n)：返回排序后的前n个元素，排序（根据RDD中数据的比较器）
rdd.top(1) //打印出：Array[Int] = Array(3) rdd.top(2) //打印出：Array[Int] = Array(3,3) rdd.top(3) //打印出：Array[Int] = Array(3,3,2)foreach()： 计算RDD中每个元素，但不返回到本地（只过一遍，不保存），一般配合print/println打印出数据
rdd.foreach(println)RDDs的血统关系图 Spark维护着RDDs之间的依赖关系和创建关系，叫做血统关系图 Spark使用血统关系图来计算每个RDD的需求和恢复丢失的数据
以上为RDDs的血统关系图示例（记录RDD从哪来的，往后又去干啥了）
惰性操作/延迟计算（Lazy Evaluation） RDD的创建和转换方法都是惰性操作，并不会立即执行 例如，当使用SparkContext的textFile方法从HDFS中读取文件时，Spark并不会马上从硬盘中读取文件，数据只有在必要时才会被加载 Spark仅仅记录了这个RDD是怎么创建的，在它上面进行操作又会创建怎样的RDD等信息，为每个RDD维护其血统信息，在需要时创建或重建RDD Spark对RDD的计算，在第一次使用action操作的时候才会执行 Spark通过内部记录metadata表，以表明transformations操作已经被响应了
缓存 回顾RDD的创建有两种方法，可以从存储系统中读取数据，也可以从现有RDD（集合）的转换操作而来
默认每次在RDDs上面进行action操作时，Spark都遍历这个调用者RDD的血统关系图，执行所有的转换来创建它重新计算RDDs 如果想要重复利用一个RDD（直接利用之前计算出的某个RDD结果），可以使用cache()/persist()
cache 把RDD存储在集群中执行者的内存中，实际上是将RDD物化在内存中
persist 是一个通用版的cache方法，通过传参的方法告知，缓存级别、缓存在哪：
移除 Spark只有在必要时才会从缓存占用的内存中移除老的RDD分区，可以调用RDD提供的unpersist方法手动移除RDD分区
创建 使用map()函数，返回key/value对 例如，包含数行数据的RDD，把每行数据的第一个单词作为keys，整行作为Value
val rdd = sc.textFile("/home/haha/helloSpark.txt") val rdd2 = rdd.map(line=>(line.split("")(0),line)) // line=>(key,value) key值：用空格分割整行，第0个数据即为第一个单词；value值：整行 //打印出： (hello,hello spark) (hello,hello spark !) //rdd2则为KeyValue对RDDTransformation操作
combineByKey()： 参数有：createCombiner,mergeValue,mergeCombiners,partitioner
此函数是基于key的聚合函数中最常使用的，返回的类型可以与输入类型不一样 许多基于key的聚合函数都用到了它，像groupByKey()
实现过程： 遍历partition中的元素，以元素的key作为判断依据 如果遍历到的是第一次出现的key，则该元素视为新元素，使用提供的createCombiner()函数初始化 如果遍历到的是一个partition中已经存在的key，使用mergeValue函数merge（整合） 合计每个partition结果时，使用mergeCombiners()函数，对所有分区处理结果进行整合
//已知Jack和Mike两人的语文、数学、英语成绩，求各自总分、平均分 val sorces = sc.parallelize(Array(("jack",80.0),("jack",90.0),("jack",85.0),("mike",85.0),("mike",92.0),("mike",94.0)) val score2 = score.combineByKey(score=>(1,score),(c1:(Int,Double),newScore)=>(c1._1+1,c1._2+newScore),(c1:(Int,Double),c2:(Int,Double))=>(c1._1+c2._1,c1._2+c2._2)) //createCombiner()：score->value值，1->计数作用; //mergeValue()：声明key/value形式的变量c1，Int->科目数，Double->科目成绩累加值，newScore->遍历到的新的分数，c1._1->c1的key值，c1._2是c1的value值（再次扫到某人，科目加1，成绩累加） //createCombiners()：c1->某个分区最终值，c2->另一个分区的最终值，分区的科目、成绩分别对应相加 score2.foreach(println) //打印出： (jack,(3,255.0)) (mike,(3,267.0)) val average = score2.map{case(name,(num,score))=>(name,score/num)} //case：判断传过来的数据类型是否匹配 average.foreach(println) //打印出： (mike,89.0) (jake,85.0)Action操作 （以键值对集合{(1,2),(3,4),(3,6)}为例）
函数描述示例结果countByKey()对每个键对应的元素分别计数rdd.countBykey(){(1,1),(3,2)}CollectAsMap()将结果以映射表的形式返回，以便查询rdd.collectAsMap()Map{(1,2),(3,6)}lookup(key)返回给定键对应的所有值rdd.lookup(3)[4,6]对于CollectAsMap()的说明：如果RDD中同一个Key中存在多个Value，那么后面的Value将会把前面的Value覆盖，最终得到的结果就是Key唯一，而且对应一个Value
参考： Spark从零开始（慕课网）
