温江做网站
新网创想网站建设，新征程启航
为企业提供网站建设、域名注册、服务器等服务
SparkCore的RDD
(1)RDD的介绍
RDD(Resilient Distributed Dataset)叫做分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变（RDD中的数据，不能增删改），可分区、元素可并行计算的集合。
具有数据流的模型的特点，自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显示的将工作集缓存在内存中。后续的查询能够重用工作集，这极大地提升了查询速度。
RDD可以从 三方面理解：
- 数据集：RDD是数据集合的抽象，是复杂物理介质上存在数据的一种逻辑视图。从外部看RDD的确可以被看待成经过封装，带扩展特性（如容错性）的数据集合。
- 分布式：RDD的数据可能存储在多个节点的磁盘上或者内存中，也就是所谓的多级存储。
- 弹性：虽然 RDD 内部存储的数据是只读的，但是，我们可以去修改（例如通 过 repartition 转换操作）并行计算计算单元的划分结构，也就是分区的数量。
总之：RDD就是一个大集合，将所有的数据都加载到内存中，方便多次进行重用。它的数据可以在多个节点上，并且RDD可以保存在内存中，当如果某个阶段的RDD丢失，不需要重新计算，只需要提取上一次的RDD，在相应的计算即可。
成都创新互联公司是一家集网站设计、成都网站建设、网站页面设计、网站优化SEO优化为一体的专业网络公司，已为成都等多地近百家企业提供网站建设服务。追求良好的浏览体验,以探求精品塑造与理念升华,设计最适合用户的网站页面。 合作只是第一步，服务才是根本,我们始终坚持讲诚信，负责任的原则，为您进行细心、贴心、认真的服务,与众多客户在蓬勃发展的市场环境中,互促共生。
(2)RDD的属性
1）A list of partitions(一组分片，数据集的基本单位)
一个分区通常与一个任务向关联，分区的个数决定了并行的粒度。分区的个数可以在创建RDD的时候指定，如果不指定，那么默认的由节点的cores个数决定。最终每一个分区会被映射成为BlockManager 中的一个Block,而这个Block会被下一个task使用进行计算。
2）A function for computing each split（算子）
每一个RDD都会实现compute，用于分区进行计算
3）A list of dependencies on other RDDs(RDD之间的依赖)
RDD 的每次转换都会生成一个新的 RDD，所以 RDD 之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark 可以通过这个依赖关系重新计算丢失的分区数据， 而不是对 RDD 的所有分区进行重新计算。
宽依赖和窄依赖：
窄依赖（完全依赖）：一个父分区唯一对应一个子分区，例：map操作
宽依赖（部分依赖）：一个父分区对应多个子分区，如：reduce、group操作
区分宽依赖和窄依赖：当前这个算子的执行过程中是否有shuffle操作。
4）Optionally a Partitioner for key-value RDDs（分区函数）
当前 Spark 中实现了两种类型的分片函数，一个是基于哈希的 HashPartitioner，另外一个是基于范围的 RangePartitioner。只有对于 key-value 的 RDD，才会有 Partitioner，非 key-value的 RDD 的 Parititioner 的值是 None。Partitioner 函数不但决定了 RDD 本身的分片数量，也决 定了 parent RDD Shuffle 输出时的分片数量。
5）Optionally a list of preferred locations to compute each split on
一个列表，存储存取每个 Partition 的优先位置（preferred location）。按照”移动数据不如移动计算”的理念，Spark 在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。而这个列表中就存放着每个分区的优先位置。
(3)RDD的API(相关算子)
RDD编程中有两种中形式：Transformation（转换）和Action（行动）。
Transformation：表示把一个RDD ---->RDD。
Action：表示把RDD----集合或者scala对象。
1)RDD的创建：
object SparktTest {
def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf()
conf.setAppName("SparktTest")
conf.setMaster("local[2]")
val sc: SparkContext = new SparkContext()
//由一个已经存在的 Scala 数据集合创建
val arr=Array(1,2,3,4)
val arr1RDD: RDD[Int] = sc.parallelize(arr)
val arr2RDD: RDD[Int] = sc.makeRDD(arr)
//由外部存储系统的数据创建（HDFS、HBase...）
val HDFSRDD: RDD[String] = sc.textFile("/data/input")
}
}2)Transformation：
官网：http://spark.apache.org/docs/latest/rdd-programming-guide.html#transformations
注意：RDD中所有的转换（Transformation）都是延迟加载，也就是说，他们并不是直接计算结果，相反的，他们只是记住这些应用到基础数据集，上的一个转换动作，只有当发生一个要求返回一个Driver动作的时候，这些转换才真正运行。
map()算子：
val HDFSRDD: RDD[String] = sc.textFile("/data/input")
/**
* map 算子，返回一个新的RDD，该RDD由每一个输入元素经过function函数转换后组成
*/
val mapRDD: RDD[(String, Int)] = HDFSRDD.map(ele=>(ele,1))flatMap()算子：
val arr=Array("hive hbase hadoop","spark hadoop","yarn hdfs")
val lineRDD: RDD[String] = sc.parallelize(arr)
/**
* flagMap：类似于map,但是每一个元素输入的元素可以被
* 映射成为0个或者多个输出的元素（返回的是一个序列，而不是单一的元素）
*/
//返回一个集合hive hbase hadoop spark hadoop yarn hdfs
val wordRDD: RDD[String] = lineRDD.flatMap(line=>line.split("\\s+"))filter()算子：
val arr=Array(1,2,3,4,5)
val arrRDD: RDD[Int] = sc.parallelize(arr)
/**
* filter过滤：返回一个新的RDD，该RDD由经过func函数计算后返回
* 值为true的输入元素组成
*/
val filterRDD: RDD[Int] = arrRDD.filter(num=>num%2==0)mapPartitions()算子：
val hdfsRDD: RDD[String] = sc.textFile("/data/input")
/**
* mapPartitions与map的唯一区别就是，mapPartitions迭代的是一个分区,
* 而map遍历的每一个元素,mapPartitions参数是一个迭代对象,返回的也是一个迭代对象
*/
val partitionRDD: RDD[String] = hdfsRDD.mapPartitions((x: Iterator[String]) => {
val temp = x.toList.map(line => line + "!")
temp.toIterator
})mapPartitionsWithIndex()算子：
val hdfsRDD: RDD[String] = sc.textFile("/data/input")
/**
* 第一个参数是分区编号：分区编号是从0开始的不间断的连续编号
* 第二个参数和mapPartitions相同
*/
val partitionRDD: RDD[String] = hdfsRDD.mapPartitionsWithIndex((parnum:Int,x: Iterator[String]) => {
println(parnum) //分区编号
val temp = x.toList.map(line => line + "!")
temp.toIterator
})sample()算子：
val list=1 to 5000
/**
* sample方法有三个参数：
* withReplacement：代表是否有放回的抽取（false 不放回，true：放回）
* fraction：抽取样本空间占总体的比例，（以分数的形式） 0<=fraction <=1
* seed:随机数生成器，new Random().nextInt(10)，不传表示使用系统的
* 注意：我们使用的sample算子，不能保证提供集合大小就恰巧是rdd.size()*fraction,结果大小会上下浮动
* sample在做抽样调查的时候，特别受用
*/
val listRDD: RDD[Int] = sc.parallelize(list)
val sampleRDD: RDD[Int] = listRDD.sample(false,0.2)
println(sampleRDD.count()) //大概是5000*0.2 上下浮动groupByKey()算子：
val list=List(("math",18),("hbase",18),("hive",22),("hive",18))
/**
* groupByKey,分组
* 建议groupByKey在实践中，能不用就不用，主要因为groupByKey的效率低，
* 因为有大量的数据在网络中传输，而且还没有进行本地的预处理
* 可以使用reduceByKey或者aggregateByKey或者combineByKey代替这个groupByKey
*/
val stuRDD: RDD[(String, Int)] = sc.parallelize(list)
//分组
val groupRDD: RDD[(String, Iterable[Int])] = stuRDD.groupByKey()
//求平均值
val result: RDD[(String, Double)] = groupRDD.map { case (name, score) => {
val avg = score.sum.toDouble / (score.size)
(name, avg)
}
}reduceByKey算子：
val list=List(("math",18),("hbase",18),("hive",22),("hive",18))
/**
* reduceByKey:在一个(K,V)对的数据集上使用，返回一个(K,V)对的数据
* 集，key 相同的值，都被使用指定的 reduce 函数聚合
* 到一起。和 groupByKey 类似，任务的个数是可以通过
* 第二个可选参数来配置的。
*/
val stuRDD: RDD[(String, Int)] = sc.parallelize(list)
//分组,求总分
val sumRDD: RDD[(String, Int)] = stuRDD.reduceByKey((x, y)=>x+y)
sumRDD.foreach(println) //打印：（hbase，36）（math，18）（hbase,18）sortByKey()算子：
val list=List(("math",18),("hbase",18),("hive",22),("hive",18))
/**
* sortByKey:在一个(K,V)的 RDD 上调用，K 必须实现 Ordered 接口，
* 返回一个按照 key 进行排序的(K,V)的 RDD
*/
//分组,求总分,排序
val stuRDD: RDD[(String, Int)] = sc.parallelize(list)
val sumRDD: RDD[(String, Int)] = stuRDD.reduceByKey((x, y)=>x+y)
sumRDD.foreach(println) //打印：（hbase，36）（math，18）（hbase,18）
val sortRDD: RDD[(String, Int)] = sumRDD.map(kv=>(kv._2,kv._1)).sortByKey().map(kv=>(kv._2,kv._1))
sortRDD.foreach(println)sortBy算子：
val list=List(("math",18),("hbase",18),("hive",22),("hive",18))
/**
* sortBy(func,[ascending], [numTasks])
* 与 sortByKey 类似，但是更灵活
* 第一个参数是根据什么排序
* 第二个是怎么排序，true 正序，false 倒序
* 第三个排序后分区数，默认与原 RDD 一样
*/
//分组,求总分,排序
val stuRDD: RDD[(String, Int)] = sc.parallelize(list)
val sumRDD: RDD[(String, Int)] = stuRDD.reduceByKey((x, y)=>x+y)
sumRDD.foreach(println) //打印：（hbase，36）（math，18）（hbase,18）
val sortRDD: RDD[(String, Int)] = sumRDD.sortBy(kv=>kv._2,false,2)aggregateByKey()算子：
object SparktTest {
def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf()
conf.setAppName("SparktTest")
conf.setMaster("local[2]")
val sc: SparkContext = new SparkContext()
/**
* aggregateByKey(zeroValue)(seqOp,combOp, [numTasks])
* 先按分区聚合再总的聚合，每次要跟初始值交流
* zeroValue:初始值
* seqOp:迭代操作，拿RDD中的每一个元素跟初始值进行合并
* combOp:分区结果的最终合并
* numTasks:分区个数
* aggregate+groupByKey=aggregateByKey
* aggregate对单个值进行RDD，aggregateByKey对（K，V）值进行RDD
*/
//aggregate
val list = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
val listRDD: RDD[Int] = sc.parallelize(list)
//求平均值
/**
* seqOp: (U, T) => U
* combOp: (U, U) => U
* u:(Int,Int) 总和，总次数
*/
val result: (Int, Int) = listRDD.aggregate(0, 0)((u: (Int, Int), x: Int) => {
(u._1 + x, u._2 + 1)
}
, (u1: (Int, Int), u2: (Int, Int)) => {
(u1._1 + u2._1, u1._2 + u2._2)
})
println(result._1 / result._2)
//aggregateByKey已经根据（k，v）k 进行分组，以下的操作，是对v进行操作
//以下操作时求平均值
val list1 = List(("math", 18), ("hbase", 18), ("hive", 22), ("hive", 18))
val stuRDD: RDD[(String, Int)] = sc.parallelize(list1)
val reslutRDD2: RDD[(String, (Int, Int))] = stuRDD.aggregateByKey((0, 0))((x: (Int, Int), y: Int) => {
(x._1 + y, x._2 + 1)
}, (x: (Int, Int), y: (Int, Int)) => {
(x._1 + y._1, x._2 + y._2)
})
reslutRDD2.foreach(kv=>{
val name=kv._1
val avg=kv._2._1.toDouble/kv._2._2
})
}
}foldLeft()算子：（不是spark的算子，是scala的高级操作）
/**
* foldLeft
* (zeroValue: T) 初值值
* (B, A) => B B是一个元组，B._1 表示累加元素，B._2 表示个数， A 表示下一个元素
*/
//aggregate
val list = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
val result: (Int, Int) = list.foldLeft((0,0))((x, y)=>{(x._1+y,x._2+1)})
println(result._1.toDouble/result._2)combineByKey()算子：
object SparktTest {
def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf()
conf.setAppName("SparktTest")
conf.setMaster("local[2]")
val sc: SparkContext = new SparkContext(conf)
/**
* combineByKey:
* 合并相同的 key 的值 rdd1.combineByKey(x => x, (a: Int,
* b: Int) => a + b, (m: Int, n: Int) => m + n)
*/
//求平均值
val list1 = List(("math", 18), ("hbase", 18), ("hive", 22), ("hive", 18))
val listRDD: RDD[(String, Int)] = sc.parallelize(list1)
/**
* createCombiner: V => C,
* mergeValue: (C, V) => C,
* mergeCombiners: (C, C) => C): RDD[(K, C)]
*/
val resultRDD: RDD[(String, (Int, Int))] = listRDD.combineByKey(x => {
(x, 1)
},
(x: (Int, Int), y: Int) => {
(x._1 + y, x._2 + 1)
},
(x: (Int, Int), y: (Int, Int)) => {
(x._1 + y._1, x._2 + y._2)
})
resultRDD.foreach{case (name,(sum,count))=>{
val avg=sum.toDouble/count
println(s"${name}:${avg}")
}}
}
}连接操作：
object SparktTest {
def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf()
conf.setAppName("SparktTest")
conf.setMaster("local[2]")
val sc: SparkContext = new SparkContext(conf)
val arr1 = Array(1, 2, 4, 5)
val arr1RDD = sc.parallelize(arr1)
val arr2 = Array(4, 5, 6, 7)
val arr2RDD = sc.parallelize(arr2)
//cartesian 笛卡尔积
val cartesianRDD: RDD[(Int, Int)] = arr1RDD.cartesian(arr2RDD)
//union : 连接
val unionRDD: RDD[Int] = arr1RDD.union(arr2RDD)
//subtract,求，差集
val sbutractRDD: RDD[Int] = arr1RDD.subtract(arr2RDD)
//join
val list1 = List(("a", 1), ("b", 2), ("c", 3))
val list1RDD = sc.parallelize(list1)
val list2 = List(("a", "zs"), ("b", "sl"))
val list2RDD = sc.parallelize(list2)
/**
* 根据元组中的key进行join 操作，相同的key向连接
* 返回的是RDD[(String, (Int, String))] （key，连接结果）
*/
val joinRDD: RDD[(String, (Int, String))] = list1RDD.join(list2RDD)
//cogroup
/**
* (String key ,
* (Iterable[Int] arr1中的相应的key所有value的集合
* , Iterable[String])) arr2中的相应的key所有value的集合
*/
val cogroupRDD: RDD[(String, (Iterable[Int], Iterable[String]))] = list1RDD.cogroup(list2RDD)
}
}分区操作：
object SparktTest {
def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf()
conf.setAppName("SparktTest")
conf.setMaster("local[2]")
val sc: SparkContext = new SparkContext(conf)
val hdfsRDD: RDD[String] = sc.textFile("/data/word.txt")
/**
* 表示在执行了filter操作之后，由于大量的数据被过滤，导致之前设定的分区task个数，
* 处理剩下的数据导致资源浪费，为了合理高效的利用资源，
* 可以对task重新定义，在coalesce方法中的分区个数一定要小于之前设置的分区个数。
*/
hdfsRDD.coalesce(2)
//打乱数据，重新分区，分区规则为随机分区
hdfsRDD.repartition(3)
//自定义分区规则(注意，只在有key-value的RDD中可以使用)
var arr1 = Array(("a", 1), ("a", 2), ("c", 1), ("b", 2), ("d", 2)
("b", 2), ("e", 2)
, ("b", 2)
, ("f", 2), ("g", 2), ("h", 2))
val arrRDD: RDD[(String, Int)] = sc.parallelize(arr1,4)
arrRDD.partitionBy(new MyPartitioner(3))
}
}
class MyPartitioner(val numPTN:Int) extends Partitioner{
//分区个数
override def numPartitions: Int = numPTN
//分区规则
override def getPartition(key: Any): Int = {
val num=key.hashCode()&Integer.MAX_VALUE%numPTN
return num
}
}总结：
- Transformation返回的仍然是一个RDD
- 它使用了链式调用的设计模式，对一个 RDD 进行计 算后，变换成另外一个 RDD，然后这个 RDD 又可以进行另外一次转换。这个过程是分布式的。
3)Action：
常见操作：
object SparktTest {
def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf()
conf.setAppName("SparktTest")
conf.setMaster("local[2]")
val sc: SparkContext = new SparkContext(conf)
val list = List(("math", 18), ("hbase", 18), ("hive", 22), ("hive", 18))
val listRDD: RDD[(String, Int)] = sc.parallelize(list)
//action rdd ---map
listRDD.reduceByKeyLocally((x,y)=>x+y)
//调用collect的目的是：触发所有的计算，最终收集当前这个调用者RDD的所有数据，返回到客户端,如果数据量比较大，谨慎使用
listRDD.collect()
//统计RDD中有多少记录
listRDD.count()
//取出RDD中的第一条记录
listRDD.first()
//取出RDD前几条记录
listRDD.take(5)
//随机采样
listRDD.takeSample(false,20)
//按照某种格式，排序后的前几条
listRDD.top(50)
//按照升序或者降序，取相应的条数的记录(其中的元素必须继承Ordered)
listRDD.takeOrdered(3)
//统计每一个key中的value有多少个
listRDD.countByKey()
//统计有多少个元素
listRDD.countByValue()
//遍历RDD中每一个元素
listRDD.foreach(kv=>{})
//分区遍历RDD中的元素
listRDD.foreachPartition(kv=>{})
//将RDD的结果，保存到相应的文件系统中(注意这个目录一定是不存在的目录)
listRDD.saveAsTextFile("/data/output")
}
}总结：Action返回值不是一个RDD。它要么是一个scala的集合，要么是一个值，要么是空。最终返回到Driver程序，或者把RDD写入到文件系统中。
当前名称：SparkCore的RDD
文章路径：http://wjwzjz.com/article/pcgcsc.html