Spark的逻辑处理流程（二）

发布于：2024-10-24 编辑：匿名来源：网络

【往期链接】：Spark的基本流程（一）前言本文参考徐立杰老师的《大数据处理框架Apache Spark设计与实现》，在此记录相关笔记，并添加一些个人理解，如有请指正我错了。参考链接：Spark逻辑处理流程概述上一章提到了逻辑处理流程（Logical plan），本章将详细讲解。

一般来说，逻辑处理流程可以分为四个部分：数据源（Data Block）：即数据从哪里来，比如 HDFS、k-y 形式的 HBase 等。数据模型：简单来说可以是理解为抽象数据类型。

例如，在MapReduce中，它是记录。 Spark对此进行了改进，这就是RDD（弹性分布式数据集）。

本质上是一个抽象类。与ArrayList等数据类型相比，RDD有两个主要区别：它只是一个逻辑概念。

简单来说，如果用户不进行cache()，RDD就不会存储在内存中，计算完成后就会消失。 ArrayList等数据结构都是驻留在内存中的。

可以包含多个分区。毕竟Spark处理的是分布式计算，必须支持多个分区。

数据操作：也就是我们可以用RDD做什么。简单来说，分为两个操作，一个是transformation()，另一个是action()：transformation()：顾名思义，就是对RDD到处改变。

这里注意，RDD本身是不能修改的，也就是说每次变换都完成了。创建一个新的，之前的不会被移动。

（结合上一点，就是说如果不使用cache()，那么在transformation()完成之前RDD就没有了）action()：是计算最终的结果。 action() 会触发 Spark 提交作业，从而真正开始执行。

例如，count() 一次和 show() 一次。（所以使用时需要注意，如果前面的RDD或者Dataframe中没有cache()，后面又重复count()和show()，就会重复计算。

）计算处理结果：通常分布式计算结果分为两种类型。：无需在Driver端计算：例如直接将结果保存在分布式文件系统（如HDFS）中。

rdd.save("").需要在Driver端计算：比如count()需要使用一个task来统计每个RDD中分区元素的数量，然后聚合到Driver中，最后相加进行计算。 2.2 Spark逻辑处理过程生成方法对于Spark来说，需要一种通用的方法，能够自动将应用转化为确定性的逻辑处理过程，即RDD之间的数据依赖关系。

因此，需要解决三个问题：如何生成R5DD，以及生成什么样的RDD？如何建立RDD之间的数据依赖关系？如何计算RDD中的数据？ 2.2.1 如何生成RDD，生成什么样的RDD？一般来说，执行Transformer操作时会产生RDD。这里有两种类似的情况：一个 Transformer 只创建一个 RDD。

比如一些简单的操作比如map(func)。一个 Transformer 创建多个 RDD。

例如，更复杂的操作如join()和distinct()，中间会生成多个RDD，但最终只会返回一个用户。下面展示了一些常用的 Transformation() 生成的 RDD： TransformationGenerated RDDsCompute()map(func)MappedRDDiterator(split).map(f)filter(func)FilteredRDDiterator(split).filter(f)flatMap(func)FlatMappedRDiterator( split).flatMap(f)mapPartitions(func)MapPartitionsRDDf(iterator(split))mapPartitionsWithIndex(func)MapPartitionsRDDf(split.index, iterator(split))sample(withReplacement,fraction,seed)PartitionwiseSampledRDDPoissonSampler.sample(iterator(split)) BernoulliSampler.sample(iterator(split))pipe(command, [envVars])PipedRDDunion(otherDataset) 生成多个 RDDintersection(otherDataset) 生成多个 RDDdistinct([numTasks])) 生成多个 RDDgroupByKey([numTasks]) 生成多个 RDDreduceByKey(func, [numTasks]) 生成多个 RDDsortByKey([ascending], [numTasks]) 生成多个 RDDjoin(otherDataset, [numTasks]) 生成多个 RDDcogroup(otherDataset, [numTasks]) 生成多个 RDDcartesian(otherDataset) 生成多个 RDDcoalesce(numPartitions) 生成多个RDDrepartition(numPartitions) 生成多个RDD2.2.2 RDD之间如何建立数据依赖关系？如上所述，transformation()之后会生成一堆RDD。

下一个问题是这些 RDD 之间的数据依赖关系是什么。一般来说可以分为两类：窄依赖（NarrowDependency）和宽依赖（ShuffleDependency）。

区分这两类依赖关系的依据是：生成的子RDD的每个分区是否完全依赖于父RDD的每个分区的全部或部分。如果依赖于每个分区的整个分区，则为窄依赖；如果仅依赖于每个分区的一部分，则为宽依赖。

划分NarrowDependency和ShuffleDependency的原因是为了生成物理执行图。 1）窄依赖（NarrowDependency）一对一依赖（OneToOneDependency）（1:1）一对一映射关系，如map()、filter()。

RangeDependency(1:1)可以理解为区域一对一，比如union()，如上图所示。多对一依赖（非官方定义，属于 NarrowDependency）（N:1）表示子 RDD 中的某个分区依赖于多个父 RDD 中的分区，例如 join()、cogroup()。

下图是 join() 的一个例子，它将两个 RDD 聚合在一起。首先执行 cogroup() 获得类型为的 MappedValuesRDD，然后将 Iterable[V1] 和 Iterable[V2] 制作为笛卡尔集合，并对其进行 flat() 操作。

多对多依赖（非官方定义，属于NarrowDependency）（N:N）是指子RDD中的一个分区依赖于多个父RDD中的分区，一个父RDD又依赖于多个子RDD。常见的是 cartesian()2) 宽依赖（ShuffleDependency）。

宽依赖也可以理解为“部分依赖”。宽依赖和MapReduce中shuffle的数据依赖是一样的（mapper对其输出进行分区，然后每个reducer对所有mapper进行分区，输出中自己的分区通过HTTP fetch获取）。

从上图可以看出，广泛依赖的子RDD的分区只会依赖于父RDD的一部分。为什么要区分窄依赖和宽依赖？主要原因是在执行过程中，窄依赖关系可以在同一阶段进行流水线化，而无需进行shuffle。

顾名思义，宽依赖需要 shuffle。另外，这样区分的话，也很容易实现。

2.2.3 如何计算RDD中的数据。上面我们讲了RDD之间的依赖关系。

现在我们来谈谈如何计算每个RDD的每个分区中的数据（记录）。 Spark的大部分transformation()函数与映射函数类似，都有固定的计算方法，称为“控制流”。

下图给出了两个例子。虽然两者在数据依赖上都具有一对一的依赖关系（OneToOneDependency），但是两者的func的“控制流程”是不同的：map(fuc)：读取一条记录，处理一条记录。

记录，然后吐出处理后的mapPartitions(fuc)：它处理分区中的所有记录，然后集中输出。

站长声明

标签：

上一篇：小样本工具1.半监督一致性正则化Temporal Ensemble & Mean Teacher代码实现

下一篇：【问答对话】kbqa？如何有知识地生成开放域问答

镁伽科技黄瑜清：智能自动化给生命科学带来巨大变革

近日，戈壁创投年度投资峰会在线上举行。戈壁创投邀请LP及被投企业经理参加会议，分享收获，共同努力。探索趋势，见证未来。 2019年，国际环境复杂严峻，资本市场波动剧烈，加上疫情影响，股权投资面临前所未有的考验。在“危机”与“机遇”交叉的环境下，戈壁创投继续保持

06-18
超星星完成数亿元C轮融资，加速释放优质碳化硅衬底产能

据投资界（ID：pedaily）12月14日消息，江苏超星星半导体股份有限公司超星行（以下简称“超星行”）近日完成数亿元C轮融资。本轮融资由国际知名投资机构领投，商洛电子、老股东策资本跟投。云秀资本担任本轮融资独家财务顾问。超星星成立于今年4月，总部位于江苏南京。致

06-18
以36.5亿元卖掉公司后，他流落街头说：我再也不会创业了，我要投资！

沉寂了五个多月后，橙晶创始人乌海因为一篇文章再次回到公众视野。　　2天前，吴海参加了摩根士丹利举办的一场“庆祝交易成功”的聚会（橙晶酒店“卖身”给华住酒店）。 “为了庆祝我的公司出售，我的心情可能不太好。我不知道为什么。醉了。 ”在《卖了酒店，昨晚，我喝醉

06-18
13小时破1207亿！砍单的背后是单打的狂欢，潮流已至

13:09:49，天猫双11全球狂欢成交额破亿元！地球已经无法阻挡国人“恐怖”的购买力了！虽然本次双11小败家的最终战斗力尚未揭晓，但与去年的亿元数据相比，已经提前了10小时50分11秒。在同时喊出“痛”和“爽”的同时，国人到底能以怎样的数字打破世人的想象，还有待观察。很

06-17
首次发布 - Gluetacs Therapeutics完成A轮融资，加速蛋白降解药物临床转化

投资界（ID：pedaily）5月31日消息，Gluetacs Therapeutics宣布获得A轮融资，由黄埔生物医药基金领投，其次是广东造币投资、南湾百奥、思南元科。本轮融资将重点关注博信生物的产品管线GT、GT的临床一期推进及临床前项目开发。标新生物是上海科技大学免疫化学研究所孵化的一

06-17
机构也“疯狂”！北京交易所成立以来，累计开展调查762次，谁是“机构调查之王”？

作者｜徐明辉编辑｜六耳源｜直达北京交流年已结束。回顾今年的经济发展，北京证券交易所是中国资本市场绕不开的话题。北京交易所作为服务创新型中小企业的主阵地，将成就一批中小企业。一些企业从被忽视，到如今已颇具规模，如今正站在聚光灯下，接受机构的深入研究。据

06-18
菜鸟驿站进军数字化社区生活，正式推出团购、洗衣、回收服务

进军团购、洗衣、回收……菜鸟驿站刚刚宣布，将从快递服务全面升级为数字化社区生活服务。据投资界（微信ID：pedaily）消息，今日（6月23日）全球智慧物流峰会上，菜鸟小站宣布升级为数字社区生活小站：通过团购将值得信赖的产品送到你家门口、洗衣、回收等便捷服务。这意味

06-17
投资界独家-传闻宝宝树引入互联网巨头加持，估值约150亿元

据投资界5月28日消息，有消息称，国内母婴龙头企业宝宝树将引入互联网+来自巨头互联网的新一轮战略投资，最新估值约为1亿元人民币。　　援引该消息，人士表示，投资合作计划将于近期公布。除了战略资本合作、进一步优化股东结构布局外，这个互联网平台也将极大赋能宝宝树在

06-17
iQOO Z3图赏：售价2000元以下的“能手卡”

不到一个月的时间，iQOO就接连发布了两款新机，有点让人应接不暇。 iQOO Neo 5和iQOO Z3都打“性价比”牌，都有一定的特色，是iQOO品牌主打销量的两条产品线。 ▲新发布的iQOO Z3。我们在体验iQOO 7时，曾说它是“三双”高手。它的存在就是带领整个队伍的进攻去攻占城市和领

06-21
普通人对亚运会的热情尽在快手

这个中秋国庆假期，没有什么话题能比杭州亚运会更火爆了。自上月14日亚运会门票开售以来，不少赛事门票都被观众抢购一空。除了观看赛事本身，看明星在亚运会上讨论比赛、为中国队加油、分享自己的观赛感受也成为一种热潮。随着29日比赛男子50米蛙泳决赛覃海洋率先冲线，中

06-18
上海港汽车出口同比增长超过50%

上海港汽车出口开门红。海通码头1月份出口各类车辆超过2万辆，同比增长超过50%。上海作为全国最大的汽车进出口口岸，正在改变过去“出口产品低端、出口市场低端”的局面。过去60%以上出口到拉美、非洲、中东等地区，到现在欧洲、美国、新西兰、澳大利亚等发达国家占比接近

06-18
硅谷精英所信奉的“AI宗教”到底是做什么的？

作者 |高念编辑|靖宇滑雪的终点是骨科，科学的终点是……神学？ 2019年是当之无愧的“AI+大模型”年。以ChatGPT为代表的生成式AI的快速进步，甚至让人们认为大型语言模型有资格被称为“世界模型”——人工智能从未像今天这样。如此接近“神性”。更难以想象的是，八年前，

06-17