文章作者邮箱:yugongshiye@sina.cn 地址:广东惠州
▲ 本章节目的
⚪ 了解Spark的RDD结构;
⚪ 掌握Spark的RDD操作方法;
⚪ 掌握Spark的RDD常用变换方法、常用执行方法;
一、Spark最核心的数据结构——RDD弹性分布式数据集
1. 概述
初学Spark时,把RDD看做是一个集合类型(类似于Array或List),用于存储数据和操作数据,但RDD和普通集合的区别:
1. RDD有分区机制,可以分布式,并行的处理同一个RDD数据集,从而极大提高处理效率。分区数量由程序员自己定。
2. RDD由容错机制。即数据丢失后,可以进行恢复。
2. 创建RDD方法
RDD就是带有分区的集合类型
弹性分布式数据集(RDD),特点是可以并行操作,并且是容错的。有两种方法可以创建RDD:
1. 执行Transform操作(变换操作)。即将一个普通集合(Array或List)转变为一个RDD。
例如:val r1 = sc.parallelize(a1,2)
或 val r1 = sc.makeRDD(List(1,2,3,4),2)
查看分区数量:r1.partitions.size。
查看分区数据:r1.glom.collect。
查看RDD整体数据:r1.collect。
2. 读取外部存储系统的数据集,如HDFS,HBase,或任何与Hadoop有关的数据源。
读取Linux本地文件:val r4 = sc.textFile("file:home/1.txt",2)
读取hds文件:val r5 = sc.textFile("hdfs://hadoop01:9000/1.txt",2)
3. RDD入门示例
案例一:
并行化集合可以通过调用 Spark Context 的并行化方法被创建,这个方法是在驱动程序(Scala-Seq)中的现有集合上的。集合里的参数会被拷贝到可以并行执行的分布式数据集里。如下例子就是如何创建一个包含了 1 到 5 的并行化集合。例如:
val data = Array(1, 2, 3, 4, 5)
val r1 = sc.parallelize(data)
val r2 = sc.parallelize(data,2)
你可以这样理解RDD:它是spark提供的一个特殊集合类。诸如普通的集合类型,如传统的Array:(1,2,3,4,5)是一个整体,但转换成RDD后,我们可以对数据进行Partition(分区)处理,这样做的目的就是为了分布式。
你可以让这个RDD有两个分区,那么有可能是这个形式:RDD(1,2) (3,4)。
这样设计的目的在于:可以进行分布式运算。
注:创建RDD的方式有多种,比如案例一中是基于一个基本的集合类型(Array)转换而来,像parallelize这样的方法还有很多,之后就会学到。此外,我们也可以在读取数据集时就创建RDD。
案例二:
Spark能够从任何基于Hadoop的存储资源,创建分布式数据集。包括本地文件系统、HDFS、Cassandra、HBase、Amazon S3等等。Spark支持TEXT文件格式、SequenceFiles文件格式和其他Hadoop的输入文件格式。
RDD的TEXT文件能够通过SparkContext的方法创建。这个方法获取一个文件的URI路径(可以是本地路径、或者是hdfs://, s3n://等),然后当作一条数据集读取其中内容。例如:
val distFile = sc.textFile("data.txt")
4. 查看RDD
scala>rdd.collect
收集rdd中的数据组成Array返回,此方法将会把分布式存储的rdd中的数据集中到一台机器中组建Array。
在生产环境下一定要慎用这个方法,容易内存溢出。
查看RDD的分区数量:
scala>rdd.partitions.size
查看RDD每个分区的元素:
scala>rdd.glom.collect
此方法会将每个分区的元素以Array形式返回。
5. 分区概念
在下图所示, 一个RDD有item1~item25个数据,共5个分区,分别在3台机器上进行处理。
此外,spark并没有原生的提供rdd的分区查看工具我们可以自己来写一个。
案例三:
import org.apache.spark.rdd.RDD
import scala.reflect.ClassTag
object su {
def debug[T: ClassTag](rdd: RDD[T]) = {
rdd.mapPartitionsWithIndex((i: Int, iter: Iterator[T]) => {
val m = scala.collection.mutable.Map[Int, List[T]]()
var list = List[T]()
while (iter.hasNext) {
list = list :+ iter.next
}
m(i) = list
m.iterator
}).collect().foreach((x: Tuple2[Int, List[T]]) => {
val i = x._1
println(s"partition:[$i]")
x._2.foreach { println }
})
}
}
二、RDD的操作
1. 概述
对于RDD的操作,总的来分有三种:
1. Transformation变化操作,特点是都是懒操作,调用后并不是马上执行,比如典型的textFile方法。此外,每当调用一次变化操作(懒操作),就会产生一个新的RDD。
2. Action执行操作,特点是会触发执行。
3. Controller控制操作。
2. 常用的变化方法(懒方法):
Transformation | Meaning |
map(func) | Return a new distributed dataset formed by passing each element of the source through a function func. 返回一个新的分布式数据集,通过函数应用于RDD每一个元素,该方法的参数是一个函数。 案例: map 将函数应用到rdd的每个元素中 val rdd = sc.makeRDD(List(1,3,5,7,9) |