Spark快速上手(2)Spark核心编程-RDD简介

Spark计算框架为适应高并发和高吞吐的数据处理需求，封装了三大数据结构，以处理不同应用：
1）RDD:弹性分布式数据集
2）累加器：分布式共享只写变量
3）广播变量：分布式共享只读变量
RDD(1)

什么是RDD

RDD(Resilient Distributed Dataset)弹性分布式数据集，为Spark中最基本的数据处理模型。
它是一个抽象类，代表弹性、不可变、可分区且其中元素可并行计算的集合。
1）弹性：多方面（存储、容错、计算、分片）
2）分布式：数据存储在集群不同节点上
3）数据集：RDD只封装计算逻辑，不保存数据
4）数据抽象：RDD需要子类实现
5）不可变：RDD封装的计算逻辑不可改变，只能产生根据原来的RDD产生新的RDD，并在其中封装计算逻辑
6）可分区、并行计算
核心属性

1）分区列表
RDD数据结构中存在分区列表，用于并行计算，是实现分布式计算的重要属性
2）分区计算函数
Spark在计算时，是使用分区函数对每一个分区进行计算
3）RDD之间依赖关系
RDD是计算模型的封装，如果需求包含多个计算模型的组合，就需要多个RDD建立依赖关系
4）分区器（option）
数据为KV类型（key-value），可以通过设定分区器自定义数据的分区
5）首选位置(option)
计算数据时，可根据计算节点状态选择节点位置进行计算
基础编程

RDD创建

在Spark中从创建RDD的方式有四种：
1）集合中创建RDD，Spark主要提供两个方法：parallelize、makeRDD
e.g.

1. val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
2. val sparkContext = new SparkContext(sparkConf)
3. val rdd1 = sparkContext.parallelize(List(1,2,3,4))
4. val rdd2 = sparkContext.makeRDD(list(1,2,3,4))
5. rdd1.collect().foreach(println)
6. rdd2.collect().foreach(println)
7. sparkContext.stop()

复制代码

makeRDD方法从底层实现看就是parallelize方法
2）从外部存储（文件）创建RDD
由外部存储系统的数据集创建RDD包括：本地的文件系统、所用Hadoop支持的数据集（譬如HDFS、HBase）
e.g.

1. val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
2. val sparkContext = new SparkContext(sparkConf)
3. val fileRDD:RDD[String] = sparkContext.textFile("input")
4. fileRDD.collect().foreach(println)
5. sparkContext.stop()

复制代码

3)从其他RDD创建
通过一个RDD运算完后，再产生新的RDD。
4）直接创建RDD(new)
使用new的方式直接构造RDD，一般由Spark框架自身使用
RDD并行度与分区

e.g.

1.   val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
2.   val sparkContext = new SparkContext(sparkConf)
3.   val dataRDD:RDD[Int] = sparkContext.makeRDD(List(1,2,3,4),4)
4.   val fileRDD:RDD[String] = sparkContext.textFile("input",2)
5.   fileRDD.collect(),foreach(println)
6.   sparkContext.stop()

复制代码

默认情况下,Spark可以将一个作业切分多个任务后，发送给Executor节点并行计算，能够并行计算的任务数量称之为并行度。这个
数量可以再构建RDD时指定。不过需要注意的是，这里并行执行的任务数量，不是指切分任务的数量


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！