Spark（三）：弹性分布式数据集（RDD）

一：Scala

    Scala 是一门现代的多范式编程语言，志在以简练、优雅及类型安全的方式来表达经常使用编程模式。它平滑地集成了面向对象和函数语言的特性。Scala 运行于 Java 平台（JVM，Java 虚拟机），并兼容现有的 Java 程序。

执行如下命令，启动spark-shell：

hadoop@master:/mysoftware/spark-1.6.1$ spark-shell

二：弹性分布式数据集（RDD）

1.RDD（Resilient Distributed Dataset，弹性分布式数据集）。   

    Spark是一个分布式计算框架， 而RDD是其对分布式内存数据的抽象，能够认为RDD就是Spark分布式算法的数据结构。而RDD之上的操做是Spark分布式算法的核心原语，由数据结构和原语设计上层算法。Spark最终会将算法翻译为DAG形式的工做流进行调度，并进行分布式任务的发布。

    RDD，它在集群中的多台机器上进行了数据分区，逻辑上能够认为是一个分布式的数组，而数组中每一个记录能够是用户自定义的任意数据结构。RDD是Spark的核心数据结构，经过RDD的依赖关系造成Spark的调度顺序，经过对RDD的操做造成了整个Spark程序。

2.RDD的建立方式

    2.1 从Hadoop文件系统（或与Hadoop兼容的其余持久化存储系统，如Hive，HBase）输出（HDFS）建立。

    2.2 从父RDD转换获得新的RDD

    2.3 经过parallelize或makeRDD将单击数据建立为分布式RDD。

scala> var textFile = sc.textFile("hdfs://192.168.226.129:9000/txt/sparkshell/sparkshell.txt");  

textFile: org.apache.spark.rdd.RDD[String] = hdfs://192.168.226.129:9000/txt/sparkshell/sparkshell.txt MapPartitionsRDD[1] at textFile at <console>:27

scala> val a = sc.parallelize(1 to 9, 3)
a: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[6] at parallelize at <console>:27

scala>

3.RDD的两种操做算子: 转换（Transformation），行动（Action）。

    3.1  转换（Transformation）：延迟计算的，也就是说从一个RDD转换生成另一个RDD的转换操做不是立刻执行，须要等到有Action操做的时候才会真正触发运算。

    3.2  行动（Action）：Action算子会触发Spark提交做业(Job),并将数据输出Spark系统。

4.RDD的重要内部属性。

    4.1  分区列表：经过分区列表能够找到一个RDD中包含的全部分区及其所在地址。

    4.2  计算每一个分片的函数：经过函数能够对每一个数据块进行RDD须要进行的用户自定义函数                                                运算。

    4.3  对父RDD的依赖列表：为了可以回溯带父RDD，为容错等提供支持。

    4.4  对 key-value pair数据类型RDD的分区器，控制分区策略和分区数。经过分区函数能够肯定数据记录在各个分区和节点上的分配，减小分布不平衡。

    4.5  每一个数据分区的地址列表（如 HDFS 上的数据块的地址）。
            若是数据有副本，则经过地址列表能够获知单个数据块的全部副本地址，为负载均衡和容错提供支持。

 

4.  Spark 计算工做流

    途中描述了Spark的输入，运行转换，输出。在运行转换中经过算子对RDD进行转换。算子RDD中定义的函数，能够对RDD中的数据进行转换和操做。

    输入：在Spark程序运行中，数据从外部数据空间（eg：HDFS）输入到Spark，数据就进入了Spark运行时数据空间，会转化为Spark中的数据块，经过BlockManager进行管理。

    运行：在Spark数据输入造成RDD后，即可以经过变换算子fliter等，对数据操做并将RDD转换为新的RDD，经过行动Action算子，触发Spark提交做业。若是数据须要服用，能够经过Cache算子，将数据缓存到内存。

    输出：程序运行结束数据会输出Spark运行时空间，存储到分布式存储中（如saveAsTextFile 输出到 HDFS）或 Scala 数据或集合中（ collect 输出到 Scala 集合，count 返回 Scala Int 型数据）。
 

    

 

 

    Spark的核心数据模型是RDD，但RDD是个抽象类，具体由各子类实现，如MappedRDD，ShuffledRDD等子类。Spark将经常使用的大数据操做都转换成为RDD的子类。

对其一些基本操做的使用：

scala> 3*7
res0: Int = 21

scala> var textFile = sc.textFile("hdfs://192.168.226.129:9000/txt/sparkshell/sparkshell.txt");  
textFile: org.apache.spark.rdd.RDD[String] = hdfs://192.168.226.129:9000/txt/sparkshell/sparkshell.txt MapPartitionsRDD[1] at textFile at <console>:27

scala> textFile.count()
res1: Long = 3                                                                  

scala> textFile.first()
res2: String = 1	spark

scala> textFile.filter(line => line.contains("berg")).count() 
res3: Long = 1

scala> textFile.filter(line => line.contains("bergs")).count() 
res4: Long = 0

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)
res5: Int = 1

scala> textFile.map(line => line.split("\t").size).reduce((a, b) => if (a > b) a else b)
res6: Int = 2

 

更多学习：

http://my.oschina.net/lgscofield/blog/497145

http://blog.csdn.net/tanggao1314/article/details/51557377

http://blog.csdn.net/yeruby/article/details/41043039

 

文献： Spark大数据分析实战。