API
应用可以通过使用Spark提供的库获得Spark集群的计算能力，这些库都是Scala编写的，但是Spark提供了面向各种语言的API，例如Scala、Python、Java等，所以可以使用以上语言进行Spark应用开发。 
Spark的API主要由两个抽象部件组成：SparkContext和RDD，应用程序通过这两个部件和Spark进行交互，连接到Spark-集群并使用相关资源。

创新互联制作网站网页找三站合一网站制作公司,专注于网页设计,网站制作、成都网站设计,网站设计,企业网站搭建,网站开发,建网站业务,680元做网站,已为1000多家服务,创新互联网站建设将一如既往的为我们的客户提供最优质的网站建设、网络营销推广服务!

1.SparkContext
是定义在Spark库中的一个类，作为Spark库的入口。包含应用程序main（）方法的Driver program通过SparkContext对象访问Spark，因为SparkContext对象表示与Spark集群的一个连接。每个Spark应用都有且只有一个激活的SparkContext类实例，如若需要新的实例，必须先让当前实例失活。 
（在shell中SparkContext已经自动创建好，就是sc）

实例化SparkContext：

val sc = new SparkContext()
1
2.RDD
RDD基础概念
弹性分布式数据集（Resilient Distributed Dataset）

并行分布在整个集群中 
把指定路径下的文本文件加载到lines这个RDD中，这个lines就是一个RDD，代表是就是整个文本文件

val lines = sc.textFile("home/haha/helloSpark.txt")
1
即使这个文件超大，分片存储在多台机器上，操作时可以直接使用RDD对整体文件进行操作

RDD是Spark分发数据和计算的基础抽象类 
例如：lines.count() 
在.count()的函数操作是在RDD数据集上的，而不是对某一具体分片

一个RDD是一个不可改变的分布式集合对象 
就lines来说，如果我们对其所代表的源文件进行了增删改操作，则相当于生成了一个新的RDD，来存放修改后的数据集

Spark中所有的计算都是通过RDD的创建、转换，操作完成的

一个RDD内部由许多partitions（分片）组成 
partitions： 
每个分片包括一部分数据，分片可在集群不同节点上计算 
分片是Spark并行处理的单元，Spark顺序的，并行的处理分片

RDD创建方法
1.把一个存在的集合传给SparkContext的parallize()方法（一般测试的时候使用这个方法）

val rdd = sc.parallelize(Array(1,2,2,1),4) 
//参数1：待并行化处理的集合；参数2：分区个数

rdd.count() //查看个数
rdd.foreach(print) //查看RDD的所有对象
//注意：每次foreach出来数值顺序会不一样，因为数据存储于4个分区，从哪个分区读取数值是随机的
1
2
3
4
5
6
7
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")
1
2
3
4
5
6
7
8
9
10
11
RDD的转换操作
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)
1
2
3
4
5
6
7
8
9
10
11
filter()：接收函数后，返回只包含满足filter()函数的元素的新RDD

val lines3 = lines.filter(word=>word.contains("hello"))

lines3.foreach(println)
//打印出：
hello
hello
1
2
3
4
5
6
7
flatMap()：对每个输入元素，输出多个输出元素，也就是将RDD中元素flat（压扁）后返回一个新的RDD

val input = sc.textFile('/home/haha/helloSpark.txt')
val lines4 = input.flatMap(line=>line.split(" ")) //用空格切分一行的单词
1
2
3
集合运算 
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中没有的
1
2
3
4
5
6
7
RDD的行动操作
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类型数据总和为9
1
2
3
4
5
6
7
8
9
collect()：遍历整个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)
1
2
3
4
5
6
foreach()： 
计算RDD中每个元素，但不返回到本地（只过一遍，不保存），一般配合print/println打印出数据

rdd.foreach(println)
1
RDD的特性
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分区

KeyValue对RDDs
创建 
使用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对RDD
1
2
3
4
5
6
7
8
//一般在测试中手动创建KeyValue对RDDs方法：

val rdd3 = sc.parallelize(Array((1,2),(3,4),(3,6)))
1
2
3
4
Transformation操作

 

 

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)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
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之SparkAPI
本文链接：http://scyanting.com/article/psdihd.html