Spark基础系列之三--RDD编程基础上

传送门
Spark实战系列之一--Spark是什么
Spark实战系列之二--什么是RDD以及RDD的常用API
Spark实战系列之三--RDD编程基础上
Spark实战系列之四--RDD编程基础下
Spark实战系列之五--键值对RDD
Spark实战系列之六--数据读写
Spark实战系列之七--综合案例
Spark基础系列之八--Spark SQL是什么
Spark基础系列之九--使用Spark SQL读写数据库
传送门

一、RDD创建

1）从文件系统中加载数据，创建RDD

• Spark采用textFile()方法来从文件系统中加载数据创建RDD
• 该方法把文件的URI作为参数，这个URI可以是：
  1>本地文件系统的地址
  2>或者是分布式文件系统HDFS的地址
  3>或者是Amazon S3的地址等等

//从本地文件系统中加载数据创建RDD
val lines = sc.textFile("file:///usr/local/spark/mycode/rdd/word.txt")

//从分布式文件系统HDFS中加载数据
//三条语句是完全等价的，可以使用其中任意一种方式
val lines = sc.textFile("hdfs://localhost:9000/user/hadoop/word.txt")
val lines = sc.textFile("/user/hadoop/word.txt")
val lines = sc.textFile("word.txt")

---

2）通过并行集合（数组），创建RDD

• 可以调用SparkContext的parallelize方法

//在集合（数组）上创建
val array = Array(1,2,3,4,5)
val rdd = sc.parallelize(array)

//在列表中创建
val list = List(1,2,3,4,5)
val rdd = sc.parallelize(list)

---

二、RDD操作

1）转换操作

• 对于RDD而言，转换得到的RDD是惰性求值的，也就是说，整个转换过程只是记录了转换的轨迹，并不会发生真正的计算，只有遇到行动操作时，才会发生真正的计算，开始从血缘关系源头开始，进行物理的转换操作。

• 典例1：filter(func)

val lines = sc.textFile(file:///usr/local/spark/mycode/rdd/word.txt)
val linesWithSpark = lines.filter(line.contains("Spark"))

---

• 典例2：map(func)

val lines = sc.textFile(file:///usr/local/spark/mycode/rdd/word.txt)
val linesWithSpark = lines.map(line.contains(" "))

---

• 典例3：flatMap(func)

val lines = sc.textFile(file:///usr/local/spark/mycode/rdd/word.txt)
val linesWithSpark = lines.flatMap(line.contains(" "))

---

• 典例4：groupByKey()

groupByKey()应用于（k，v）键值对的数据集时，返回一个新的（k，Iterable）形式的数据集

---

• 典例5：reduceByKey(func)

在典例4的基础上，把每个key传递到函数func中进行聚合后得到value的结果

---

2）行动操作

• 行动操作是真正触发计算的地方。Spark程序执行到行动操作时，才会执行真正的计算，从文件中加载数据，完成一次又一次转换操作，最终，完成行动操作得到结果。

• 典例：

3）惰性机制

• 所谓的“惰性机制”是指，整个转换过程只是记录了转换的轨迹，并不会发生真正的计算，只有遇到行动操作时，才会触发“从头到尾”的真正的计算。
• 也正因为这样，对于迭代计算而言，代价是很大的，迭代计算经常需要多次重复使用同一组数据
• 可以通过持久化（缓存）机制避免这种开销，我将在下一篇Spark实战系列之四--RDD编程基础中详细说明。