网站做移动适配,网站开发转码手机,小微企业所得税优惠政策,isp网站接入做哪些业务MapReduce的缺陷
MR虽然在编程接口的种类和丰富程度上已经比较完善了#xff0c;但这些系统普遍都缺乏操作分布式内存的接口抽象#xff0c;导致很多应用在性能上非常低效 。 这些应用的共同特点是需要在多个并行操 作之间重用工作数据集 #xff0c;典型的场景就是机器学习…MapReduce的缺陷
MR虽然在编程接口的种类和丰富程度上已经比较完善了但这些系统普遍都缺乏操作分布式内存的接口抽象导致很多应用在性能上非常低效 。 这些应用的共同特点是需要在多个并行操 作之间重用工作数据集 典型的场景就是机器学习和图应用中常用的迭代算法 (每一步对数据 执行相似的函数) 。
RDD
RDD是只读的。
RDD五大属性①分区、②依赖、③计算函数、④分区器、⑤首选运行位置。
RDD 则是直接在编程接口层面提供了一种高度受限的共享内存模型如图下图所示。 RDD 是 Spark 的核心数据结构全称是弹性分布式数据集 (Resilient Distributed Dataset)其本质是一种分布式的内存抽象表示一个只读的数据分区( Partition)集合 。一个 RDD 通常只能通过其他的 RDD转换而创建。 RDD 定义了各种丰富的转换操作(如 map、 join和 filter等)通过这些转换操作新的 RDD 包含了如何从其他 RDD 衍生所必需的信息这些信息构成了 RDD 之间的依赖关系( Dependency) 。 依赖具体分为两种 一种是窄依赖 RDD 之间分区是一一对应的;另一种是宽依赖下游 RDD 的每个分区与上游 RDD (也称之为父 RDD)的每个分区都有关是多对多的关系 。 窄依赖中的所有转换操作可以通过类似管道(Pipeline)的方式全部执行宽依赖意味着数据需要在不同节点之间 Shuffle 传输 。 RDD计算的时候会通过一个 compute函数得到每个分区的数据。 若 RDD是通过已有的文件系统构建的则 compute 函数读取指定文件系统中的数据如果 RDD 是通过其他 RDD 转换而来的则 compute 函数执行转换逻辑将其他 RDD 的数据进行转换。 RDD 的操作算子包括两 类 一类是 transformation用来将 RDD 进行转换构建 RDD 的依赖关系另一类称为 action, 用来触发 RDD 的计算得到 RDD 的相关计算结果或将 RDD 保存到文件系统中。
在 Spark 中 RDD 可以创建为对象 通过对象上的各种方法调用来对 RDD 进行转换 。 经过一系列的 transformation逻辑之后就可以调用 action来触发 RDD 的最终计算。 通常来讲 action 包括多种方式可以 是 向应用程序返回结果( show、 count 和 collect等)也可以是向存 储系统保存数据(saveAsTextFile等)。 在Spark中只有遇到 action才会真正地执行 RDD 的计算(注:这被称为惰性计算英文为 LazyEvqluation)这样在运行时可以通过管道的方式传输多个转换 。
总结而言基于 RDD 的计算任务可描述为从稳定的物理存储(如分布式文件系统 HDFS) 中加载记录记录被传入由一组确定性操作构成的 DAG (有向无环图)然后写回稳定存储。 RDD还可以将数据集缓存到内存中使得在多个操作之间可以很方便地重用数据集。 总的来讲RDD 能够很方便地支持 MapReduce 应用、关系型数据处理、流式数据处理(Stream Processing) 和迭代型应用(图计算、机器学习等)。
在容错性方面基于 RDD 之间的依赖 一个任务流可以描述为 DAG。 在实际执行的时候 RDD 通过 Lineage 信息(血缘关系)来完成容错即使出现数据分区丢失也可以通过 Lineage 信息重建分区。 如果在应用程序中多次使用同一个 RDD则可以将这个 RDD 缓存起来该 RDD 只有在第一次计算的时候会根据 Lineage 信息得到分区的数据在后续其他地方用到这个 RDD 的时候会直接从缓存处读取而不用再根据 Lineage信息计算通过重用达到提升性能的目的 。 虽然 RDD 的 Lineage 信息可以天然地实现容错(当 RDD 的某个分区数据计算失败或丢 失时可以通过 Lineage信息重建)但是对于长时间迭代型应用来说随着迭代的进行RDD 与 RDD之间的 Lineage信息会越来越长一旦在后续迭代过程中出错就需要通过非常长的 Lineage信息去重建对性能产生很大的影响。 为此RDD 支持用 checkpoint机制将数据保存到持久化的存储中这样就可以切断之前的 Lineage信息因为 checkpoint后的 RDD不再需要知道它的父 RDD 可以从 checkpoint 处获取数据。
DAG
顾名思义DAG 是一种“图”图计算模型的应用由来已久早在上个世纪就被应用于数据库系统Graph databases的实现中。任何一个图都包含两种基本元素节点Vertex和边Edge节点通常用于表示实体而边则代表实体间的关系。
DAG有向无环图Directed Acyclic Graph的缩写常用于建模。Spark中使用DAG对RDD的关系进行建模描述了RDD的依赖关系这种关系也被称之为lineageRDD的依赖关系使用Dependency维护参考Spark RDD之DependencyDAG在Spark中的对应的实现为DAGScheduler。 基础概念 介绍DAGScheduler中的一些概念有助于理解后续流程。
Job调用RDD的一个action如count即触发一个Jobspark中对应实现为ActiveJobDAGScheduler中使用集合activeJobs和jobIdToActiveJob维护JobStage代表一个Job的DAG会在发生shuffle处被切分切分后每一个部分即为一个StageStage实现分为ShuffleMapStage和ResultStage一个Job切分的结果是0个或多个ShuffleMapStage加一个ResultStageTaskSet一组TaskTask最终被发送到Executor执行的任务和stage的ShuffleMapStage和ResultStage对应其实现分为ShuffleMapTask和ResultTask
把 DAG 图反向解析成多个阶段每个阶段中包含多个任务每个任务会被任务调度器分发给工作节点上的 Executor 上执行。 Web UI上DAG举例 Checkpoint
RDD的依赖
checkpoint先了解一下RDD的依赖比如计算wordcount
scala sc.textFile(hdfs://leen:8020/user/hive/warehouse/tools.db/cde_prd).flatMap(_.split(\\\t)).map((_,1)).reduceByKey(__);
res0: org.apache.spark.rdd.RDD[(String, Int)] ShuffledRDD[4] at reduceByKey at console:28scala res0.toDebugString
res1: String
(2) ShuffledRDD[4] at reduceByKey at console:28 []-(2) MapPartitionsRDD[3] at map at console:28 []| MapPartitionsRDD[2] at flatMap at console:28 []| hdfs://leen:8020/user/hive/warehouse/tools.db/cde_prd MapPartitionsRDD[1] at textFile at console:28 []| hdfs://leen:8020/user/hive/warehouse/tools.db/cde_prd HadoopRDD[0] at textFile at console:28 []
1、在textFile读取hdfs的时候就会先创建一个HadoopRDD其中这个RDD是去读取hdfs的数据key为偏移量value为一行数据因为通常来讲偏移量没有太大的作用所以然后会将HadoopRDD转化为MapPartitionsRDD这个RDD只保留了hdfs的数据。 2、flatMap 产生一个RDD MapPartitionsRDD 3、map 产生一个RDD MapPartitionsRDD 4、reduceByKey 产生一个RDD ShuffledRDD
如何建立checkPoint
1、首先需要用sparkContext设置hdfs的checkpoint的目录如果不设置使用checkpoint会抛出异常
scala res0.checkpoint
org.apache.spark.SparkException: Checkpoint directory has not been set in the SparkContextscala sc.setCheckpointDir(hdfs://leen:8020/checkPointDir)
sc.setCheckpointDir(hdfs://leen:8020/checkPointDir) 执行了上面的代码,hdfs里面会创建一个目录: /checkPointDir/9ae90c62-a7ff-442a-bbf0-e5c8cdd7982d
2、然后执行checkpoint
scala res0.checkpoint
1
发现hdfs中还是没有数据说明checkpoint也是个transformation的算子。
scala res0.count()
INFO ReliableRDDCheckpointData: Done checkpointing RDD 4 to hdfs://leen:8020/checkPointDir/9ae90c62-a7ff-442a-bbf0-e5c8cdd7982d/rdd-4, new parent is RDD 5
res5: Long 73689
1
2
3
hive dfs -du -h /checkPointDir/9ae90c62-a7ff-442a-bbf0-e5c8cdd7982d/rdd-4;
147 147 /checkPointDir/9ae90c62-a7ff-442a-bbf0-e5c8cdd7982d/rdd-4/_partitioner
1.2 M 1.2 M /checkPointDir/9ae90c62-a7ff-442a-bbf0-e5c8cdd7982d/rdd-4/part-00000
1.2 M 1.2 M /checkPointDir/9ae90c62-a7ff-442a-bbf0-e5c8cdd7982d/rdd-4/part-00001
但是执行的时候相当于走了两次流程前面计算了一遍然后checkpoint又会计算一次所以一般我们先进行cache然后做checkpoint就会只走一次流程checkpoint的时候就会从刚cache到内存中取数据写入hdfs中如下
rdd.cache()
rdd.checkpoint()
rdd.collect
在源码中在checkpoint的时候强烈建议先进行cache并且当你checkpoint执行成功了那么前面所有的RDD依赖都会被销毁如下 /*** Mark this RDD for checkpointing. It will be saved to a file inside the checkpoint* directory set with SparkContext#setCheckpointDir and all references to its parent* RDDs will be removed. This function must be called before any job has been* executed on this RDD. It is strongly recommended that this RDD is persisted in* memory, otherwise saving it on a file will require recomputation.*/def checkpoint(): Unit RDDCheckpointData.synchronized {// NOTE: we use a global lock here due to complexities downstream with ensuring// children RDD partitions point to the correct parent partitions. In the future// we should revisit this consideration.if (context.checkpointDir.isEmpty) {throw new SparkException(Checkpoint directory has not been set in the SparkContext)} else if (checkpointData.isEmpty) {checkpointData Some(new ReliableRDDCheckpointData(this))}}
RDD依赖被销毁
scala res0.toDebugString
res6: String
(2) ShuffledRDD[4] at reduceByKey at console:28 []| ReliableCheckpointRDD[5] at count at console:30 []
