Spark核心编程的三大数据结构 之 RDD基础编程 (二)
这是我参与更文挑战的第2天，活动详情查看：更文挑战数据库
前传
Spark核心编程的三大数据结构 之 RDD基础编程 (一)apache
4. RDD依赖关系
4.1 RDD 血缘关系
RDD只支持粗粒度转换，即在大量记录上执行的单个操做。将建立RDD的一系列Lineage（血统）记录下来，以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它能够根据这些信息来从新运算和恢复丢失的数据分区。编程
def main(args: Array[String]): Unit = {
val sc = new SparkContext(
new SparkConf().setMaster("local[*]").setAppName("MapPartitions")
)
val rdd = sc.makeRDD(List(
("a", 1), ("a", 2), ("b", 3), ("b", 3), ("b", 3),
("b", 4), ("b", 5), ("a", 6)
), 2)
println(rdd.toDebugString)
println("--------------------------------------")
val rdd1 = rdd.map(t => (t._1, t._2 * 2))
println(rdd1.toDebugString)
println("--------------------------------------")
val rdd2 = rdd1.mapValues(_ + 100)
println(rdd2.toDebugString)
println("--------------------------------------")
val rdd3 = rdd2.reduceByKey(_ + _)
println(rdd3.toDebugString)
println("--------------------------------------")
val res = rdd2.collect()
println(res.mkString("\n"))
}
复制代码
(2) ParallelCollectionRDD[0] at makeRDD at _1.scala:19 []
--------------------------------------
(2) MapPartitionsRDD[1] at map at _1.scala:27 []
| ParallelCollectionRDD[0] at makeRDD at _1.scala:19 []
--------------------------------------
(2) MapPartitionsRDD[2] at mapValues at _1.scala:32 []
| MapPartitionsRDD[1] at map at _1.scala:27 []
| ParallelCollectionRDD[0] at makeRDD at _1.scala:19 []
--------------------------------------
(2) ShuffledRDD[3] at reduceByKey at _1.scala:37 []
+-(2) MapPartitionsRDD[2] at mapValues at _1.scala:32 []
| MapPartitionsRDD[1] at map at _1.scala:27 []
| ParallelCollectionRDD[0] at makeRDD at _1.scala:19 []
--------------------------------------
复制代码
4.2 RDD 依赖关系
这里所谓的依赖关系，其实就是两个相邻RDD之间的关系缓存
def main(args: Array[String]): Unit = {
val sc = new SparkContext(
new SparkConf().setMaster("local[*]").setAppName("MapPartitions")
)
val rdd = sc.makeRDD(List(
("a", 1), ("a", 2), ("b", 3), ("b", 3), ("b", 3),
("b", 4), ("b", 5), ("a", 6)
), 2)
println(rdd.dependencies)
println("--------------------------------------")
val rdd1 = rdd.map(t => (t._1, t._2 * 2))
println(rdd1.dependencies)
println("--------------------------------------")
val rdd2 = rdd1.mapValues(_ + 100)
println(rdd2.dependencies)
println("--------------------------------------")
val rdd3 = rdd2.reduceByKey(_ + _)
println(rdd3.dependencies)
println("--------------------------------------")
val res = rdd2.collect()
println(res.mkString("\n"))
}
复制代码
List()
--------------------------------------
List(org.apache.spark.OneToOneDependency@38704ff0)
--------------------------------------
List(org.apache.spark.OneToOneDependency@44de94c3)
--------------------------------------
List(org.apache.spark.ShuffleDependency@2c58dcb1)
--------------------------------------
复制代码
4.3 RDD 窄依赖
窄依赖表示每个父(上游)RDD的Partition最多被子（下游）RDD的一个Partition使用.markdown
@DeveloperApi
class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd) {
override def getParents(partitionId: Int): List[Int] = List(partitionId)
}
复制代码
4.4 RDD 宽依赖
宽依赖表示同一个父（上游）RDD的Partition被多个子（下游）RDD的Partition依赖，会引发Shuffle.数据结构
@DeveloperApi
class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag]( @transient private val _rdd: RDD[_ <: Product2[K, V]], val partitioner: Partitioner, val serializer: Serializer = SparkEnv.get.serializer, val keyOrdering: Option[Ordering[K]] = None, val aggregator: Option[Aggregator[K, V, C]] = None, val mapSideCombine: Boolean = false, val shuffleWriterProcessor: ShuffleWriteProcessor = new ShuffleWriteProcessor)
extends Dependency[Product2[K, V]] {
if (mapSideCombine) {
require(aggregator.isDefined, "Map-side combine without Aggregator specified!")
}
override def rdd: RDD[Product2[K, V]] = _rdd.asInstanceOf[RDD[Product2[K, V]]]
private[spark] val keyClassName: String = reflect.classTag[K].runtimeClass.getName
private[spark] val valueClassName: String = reflect.classTag[V].runtimeClass.getName
private[spark] val combinerClassName: Option[String] =
Option(reflect.classTag[C]).map(_.runtimeClass.getName)
val shuffleId: Int = _rdd.context.newShuffleId()
val shuffleHandle: ShuffleHandle = _rdd.context.env.shuffleManager.registerShuffle(
shuffleId, this)
_rdd.sparkContext.cleaner.foreach(_.registerShuffleForCleanup(this))
_rdd.sparkContext.shuffleDriverComponents.registerShuffle(shuffleId)
}
复制代码
4.5 RDD 阶段划分
DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，该图形具备方向，不会闭环。例如，DAG记录了RDD的转换过程和任务的阶段。app
private[scheduler] def handleJobSubmitted(jobId: Int,
finalRDD: RDD[_],
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
callSite: CallSite,
listener: JobListener,
properties: Properties): Unit = {
var finalStage: ResultStage = null
try {
//建立新阶段可能会抛出异常，例如，做业运行在
//删除底层HDFS文件的HadoopRDD。
finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite)
} catch {
.......
}
/** * 建立一个与提供的jobId相关联的ResultStage */
private def createResultStage(
rdd: RDD[_],
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
jobId: Int,
callSite: CallSite): ResultStage = {
checkBarrierStageWithDynamicAllocation(rdd)
checkBarrierStageWithNumSlots(rdd)
checkBarrierStageWithRDDChainPattern(rdd, partitions.toSet.size)
val parents = getOrCreateParentStages(rdd, jobId)
val id = nextStageId.getAndIncrement()
val stage = new ResultStage(id, rdd, func, partitions, parents, jobId, callSite)
stageIdToStage(id) = stage
updateJobIdStageIdMaps(jobId, stage)
stage
}
/** * 获取或建立给定RDD的父阶段列表。新的stage将使用提供的firstJobId建立 */
private def getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
getShuffleDependencies(rdd).map { shuffleDep =>
getOrCreateShuffleMapStage(shuffleDep, firstJobId)
}.toList
}
/** * 若是在shuffleIdToMapStage中存在shuffle map stage，则获取一个shuffle map stage。不然，若是洗牌地图阶段不存在，该方法将建立洗牌地图阶段，以及任何丢失的祖先洗牌地图阶段。 */
private def getOrCreateShuffleMapStage(
shuffleDep: ShuffleDependency[_, _, _],
firstJobId: Int): ShuffleMapStage = {
shuffleIdToMapStage.get(shuffleDep.shuffleId) match {
case Some(stage) =>
stage
case None =>
//为全部缺失的祖先洗牌依赖建立阶段。
getMissingAncestorShuffleDependencies(shuffleDep.rdd).foreach { dep =>
//即便getmissing祖宗shuffledependencies只返回shuffle依赖
//没有在shuffleIdToMapStage中，当咱们
//在foreach循环中获取一个特定的依赖，它被添加到
// shuffleIdToMapStage经过早期依赖的阶段建立过程。看到
// SPARK-13902获取更多信息。
if (!shuffleIdToMapStage.contains(dep.shuffleId)) {
createShuffleMapStage(dep, firstJobId)
}
}
//最后，为给定的shuffle依赖建立一个stage。
createShuffleMapStage(shuffleDep, firstJobId)
}
}
复制代码
4.6 RDD 任务划分
RDD任务切分中间分为：Application、Job、Stage和Taskide
- Application：初始化一个SparkContext即生成一个Application；
- Job：一个Action算子就会生成一个Job；
- Stage：Stage等于宽依赖(ShuffleDependency)的个数加1；
- Task：一个Stage阶段中，最后一个RDD的分区个数就是Task的个数。
注意：Application->Job->Stage->Task每一层都是1对n的关系函数
//划分源码
val tasks: Seq[Task[_]] = try {
val serializedTaskMetrics = closureSerializer.serialize(stage.latestInfo.taskMetrics).array()
stage match {
case stage: ShuffleMapStage =>
stage.pendingPartitions.clear()
partitionsToCompute.map { id =>
val locs = taskIdToLocations(id)
val part = partitions(id)
stage.pendingPartitions += id
new ShuffleMapTask(stage.id, stage.latestInfo.attemptNumber,
taskBinary, part, locs, properties, serializedTaskMetrics, Option(jobId),
Option(sc.applicationId), sc.applicationAttemptId, stage.rdd.isBarrier())
}
case stage: ResultStage =>
partitionsToCompute.map { id =>
val p: Int = stage.partitions(id)
val part = partitions(p)
val locs = taskIdToLocations(id)
new ResultTask(stage.id, stage.latestInfo.attemptNumber,
taskBinary, part, locs, id, properties, serializedTaskMetrics,
Option(jobId), Option(sc.applicationId), sc.applicationAttemptId,
stage.rdd.isBarrier())
}
}
} catch {
case NonFatal(e) =>
abortStage(stage, s"Task creation failed: $e\n${Utils.exceptionString(e)}", Some(e))
runningStages -= stage
return
}
}
// Figure out the indexes of partition ids to compute.
val partitionsToCompute: Seq[Int] = stage.findMissingPartitions()
//findMissingPartitions 有两个实现类
//ShuffleMapStage实现
override def findMissingPartitions(): Seq[Int] = {
mapOutputTrackerMaster
.findMissingPartitions(shuffleDep.shuffleId)
.getOrElse(0 until numPartitions)
//ResultStage实现
override def findMissingPartitions(): Seq[Int] = {
val job = activeJob.get
(0 until job.numPartitions).filter(id => !job.finished(id))
}
复制代码
5.RDD持久化
5.1 RDD Cache缓存
- RDD经过Cache或者Persist方法将前面的计算结果缓存，默认状况下会把数据以缓存在JVM的堆内存中。可是并非这两个方法被调用时当即缓存，而是触发后面的action算子时，该RDD将会被缓存在计算节点的内存中，并供后面重用。
- 缓存有可能丢失，或者存储于内存的数据因为内存不足而被删除，RDD的缓存容错机制保证了即便缓存丢失也能保证计算的正确执行。经过基于RDD的一系列转换，丢失的数据会被重算，因为RDD的各个Partition是相对独立的，所以只须要计算丢失的部分便可，并不须要重算所有Partition。
- Spark会自动对一些Shuffle操做的中间数据作持久化操做(好比：reduceByKey)。这样作的目的是为了当一个节点Shuffle失败了避免从新计算整个输入。可是，在实际使用的时候，若是想重用数据，仍然建议调用persist或cache。
def main(args: Array[String]): Unit = {
val sc = new SparkContext(
new SparkConf().setMaster("local[*]").setAppName("MapPartitions")
)
val rdd = sc.makeRDD(List(
1, 2, 5, 9
), 2)
val mapRdd = rdd.map(i => {
println("map--------")
("a", i)
})
mapRdd.cache() //源码调用的persist()
//指定存储级别
//mapRdd.persist(StorageLevel.DISK_ONLY)
mapRdd.reduceByKey(_ + _).collect().foreach(println)
mapRdd.groupByKey().collect().foreach(println)
sc.stop()
}
复制代码
StorageLevel的全部存储级别
val NONE = new StorageLevel(false, false, false, false)
val DISK_ONLY = new StorageLevel(true, false, false, false)
val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
val MEMORY_ONLY = new StorageLevel(false, true, false, true)
val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
复制代码
5.2 RDD CheckPoint检查点
- 所谓的检查点其实就是经过将RDD中间结果写入磁盘
- 因为血缘依赖过长会形成容错成本太高，这样就不如在中间阶段作检查点容错，若是检查点以后有节点出现问题，能够从检查点开始重作血缘，减小了开销。
- 对RDD进行checkpoint操做并不会立刻被执行，必须执行Action操做才能触发。
def main(args: Array[String]): Unit = {
val sc = new SparkContext(
new SparkConf().setMaster("local[*]").setAppName("MapPartitions")
)
//设置检查点存储路径
sc.setCheckpointDir("./check")
val rdd = sc.makeRDD(List(
1, 2, 5, 9
), 2)
val mapRdd = rdd.map(i => {
println("map--------")
("a", i)
})
mapRdd.cache()
// checkpoint会把前面的RDD执行两次 配合cache()便可只执行一次
mapRdd.checkpoint()
mapRdd.reduceByKey(_ + _).collect().foreach(println)
mapRdd.groupByKey().collect().foreach(println)
sc.stop()
}
复制代码
5.3 缓存和检查点区别
- Cache缓存只是将数据保存起来，不切断血缘依赖。Checkpoint检查点切断血缘依赖。
- Cache缓存的数据一般存储在磁盘、内存等地方，可靠性低。Checkpoint的数据一般存储在HDFS等容错、高可用的文件系统，可靠性高。
- 建议对checkpoint()的RDD使用Cache缓存，这样checkpoint的job只需从Cache缓存中读取数据便可，不然须要再从头计算一次RDD。
6.RDD分区器
Spark目前支持Hash分区和Range分区，和用户自定义分区。Hash分区为当前的默认分区。分区器直接决定了RDD中分区的个数、RDD中每条数据通过Shuffle后进入哪一个分区，进而决定了Reduce的个数。oop
- 只有Key-Value类型的RDD才有分区器，非Key-Value类型的RDD分区的值是None
- 每一个RDD的分区ID范围：0 ~ (numPartitions - 1)，决定这个值是属于那个分区的。
6.1 Hash分区：对于给定的key，计算其hashCode,并除以分区个数取余
HashPartitioner具体代码可自行查看
6.2 Range分区：将必定范围内的数据映射到一个分区中，尽可能保证每一个分区数据均匀，并且分区间有序
RangePartitioner具体代码可自行查看
6.3 自定义分区
def main(args: Array[String]): Unit = {
val sc = new SparkContext(
new SparkConf().setMaster("local[*]").setAppName("MapPartitions")
)
val rdd = sc.makeRDD(List(
("a", 1), ("a", 2), ("b", 3), ("b", 3), ("b", 3),
("b", 4), ("b", 5), ("a", 6), ("c", 6), ("d", 6)
),5 )
val parRDD = rdd.partitionBy(new MyPartitioner)
val reduceRdd1 = parRDD.reduceByKey((i,j)=>{
println("reduceRdd1")
i+j
})
val reduceRdd2 = reduceRdd1.reduceByKey((i,j)=>{
println("reduceRdd2")
i+j
})
reduceRdd2.saveAsTextFile("out1")
sc.stop()
}
/** * 实现把 * a 分到0区 * b 分到1区 * 其余 分到2区 */
class MyPartitioner extends Partitioner {
/** * numPartitions 和 getPartition 是必须重写的方法 */
override def numPartitions: Int = 3
override def getPartition(key: Any): Int = key match {
case "a" => 0
case "b" => 1
case _ => 2
}
/** * equals 和 hashCode 可选 */
override def equals(other: Any): Boolean = other match {
case h: HashPartitioner =>
h.numPartitions == numPartitions
case _ =>
false
}
override def hashCode: Int = numPartitions
}
复制代码
7.RDD文件读取与保存
Spark的数据读取及数据保存能够从两个维度来做区分：文件格式以及文件系统。 文件格式分为：text文件、csv文件、sequence文件以及Object文件； 文件系统分为：本地文件系统、HDFS、HBASE以及数据库。
7.1 text文件
def main(args: Array[String]): Unit = {
val sc = new SparkContext(
new SparkConf().setMaster("local[*]").setAppName("MapPartitions")
)
val rdd = sc.makeRDD(List(
("a", 1), ("a", 2), ("b", 3), ("b", 3), ("b", 3)
), 1)
/** * text文件 */
rdd.saveAsTextFile("text")
sc.textFile("text").collect().foreach(println)
sc.stop()
}
复制代码
7.2 sequence文件
SequenceFile文件是Hadoop用来存储二进制形式的key-value对而设计的一种平面文件(Flat File)。在SparkContext中，能够调用sequenceFile[keyClass, valueClass](path)。
def main(args: Array[String]): Unit = {
val sc = new SparkContext(
new SparkConf().setMaster("local[*]").setAppName("MapPartitions")
)
val rdd = sc.makeRDD(List(
("a", 1), ("a", 2), ("b", 3), ("b", 3), ("b", 3)
), 1)
/** * Sequence文件 */
rdd.saveAsSequenceFile("sequence")
sc.sequenceFile[String, Int]("sequence").collect().foreach(println)
sc.stop()
}
复制代码
7.3 object对象文件
对象文件是将对象序列化后保存的文件，采用Java的序列化机制。能够经过objectFile[T: ClassTag](path)函数接收一个路径，读取对象文件，返回对应的RDD，也能够经过调用saveAsObjectFile()实现对对象文件的输出。由于是序列化因此要指定类型。
def main(args: Array[String]): Unit = {
val sc = new SparkContext(
new SparkConf().setMaster("local[*]").setAppName("MapPartitions")
)
val rdd = sc.makeRDD(List(
("a", 1), ("a", 2), ("b", 3), ("b", 3), ("b", 3)
), 1)
/** * object文件 */
rdd.saveAsObjectFile("object")
sc.objectFile[(String, Int)]("object").collect().foreach(println)
sc.stop()
}
复制代码
相关文章
- 1. Spark核心编程的三大数据结构 之 RDD基础编程 (一)
- 2. Spark核心 2 RDD编程
- 3. 17、Spark核心编程之Spark基本工作原理与RDD
- 4. Spark核心编程-RDD概述
- 5. spark核心-RDD编程指导
- 6. Spark（三）Spark RDD编程
- 7. Spark核心编程-RDD行动操做
- 8. spark核心编程，spark基本工作原理与RDD
- 9. Spark快速大数据分析之第三章RDD编程
- 10. spark RDD编程
- 更多相关文章...
- • Rust 并发编程 - RUST 教程
- • C# 程序结构 - C#教程
- • Flink 数据传输及反压详解
- • TiDB 在摩拜单车在线数据业务的应用和实践
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
欢迎关注本站公众号,获取更多信息
