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？如何有知识地生成开放域问答

7.89%！美国商务部下调我国光伏企业反倾销税率

9月，美国商务部再次更新部分中国光伏企业年度反倾销税率，由5月份公布的11.32%至 7.89%。其中，台湾联合再生能源的光伏组件税率仅为1.27%。复审期（POR）为2019年2月1日至1月31日。反倾销税率调整的企业包括英利（英利旗下共9家子公司）、嘉盛、鲍威尔特、EEPV、京瓷、台

06-08
社保基金去年四季度增持21股，增持20股

据Wind统计，截至3月28日，截至四季度末，社保基金曾出现在前十大流通股东中的74只股票。其中，新增21个，增加20个。在社保基金重仓股票中，从前十大流通股东名单中社保基金数量来看，中国巨石社保基金数量最多。前十大流通股东名单中共有3家社保基金出现，持股总数为1.32

06-17
“科技中国”企业云课堂特邀专家演讲《企业质量改善与创新文化推进》

7月6日上午，由中国科协、中华全国工商业联合会、学术部主办。中国科协、中华全国工商业联合会宣传部、经济部、中科协企业创新服务中心主办的“科技创新中国”企业云课堂科技之声继续开启，邀请了广东省卓越质量促进中心的创始人、国家创新方法三级认证的专家、高级工程师刘

06-18
手机振动体验好不好不仅仅和电机有关

三月是新手机浪潮。各家手机厂商纷纷发布了自己的旗舰手机。年初，我体验了很多旗舰甚至游戏手机。最后，我还发现了一些有趣的现象。大多数手机厂商都会默认配置横向线性马达，这几乎已经成为旗舰手机的标配硬件。当然，有营销方面的原因，但更值得注意的是，手机厂商确

06-21
中芯国际IPO后计划发行2,465万股，财务问题亟待解决

南京中芯特种金属装备有限公司首次申请于12月20日获委员批准中国证监会创业板发行审核。公司主营业务为研发特种金属压力容器的开发、设计、制造和销售。本次拟发行10,000股，发行后总股本为10,000股。　　招股书显示，中芯国际受益于装备制造业的产业支持政策，并间接撬动

06-17
东方启音获2500万美元B轮融资，专注言语治疗和智能发展

据投资界12月19日消息，国内首家致力于言语治疗的机构东方启音言语治疗（中国）有限公司治疗，宣布获得北极光创投投资，斯道资本（原富达亚洲创投）和F-Prime基金（原富达生物科技基金）联合领投，长岭资本参与1万美元B轮融资。其中，斯道资本和F-Prime基金也是东方启音的A

06-17
财政部：符合条件的集成电路企业5年免征企业所得税

关于集成电路生产企业企业所得税政策问题的通知财税【财税】27号各省财政厅（局）各自治区、直辖市、计划单列市、国家税务局、地方税务局、发展改革委、工业和信息化厅、新疆生产建设兵团财务局、发展改革委工业和信息化委员会：为进一步支持集成电路产业发展，现就有关企业

06-06
中美贸易战对全球50%的半导体造成巨大影响

第一波贸易战没有影响世界信息产业链，但第二波之后对价值100的产品加征关税亿美元，包括信息通信产品，以及全球信息通信产业都将受到影响。如果事件发展到这个地步，将会影响全球供应链。如果成本变高，售价将不可避免地上涨，而需求将减少。中美贸易战将对50%的半导体贸

06-06
韩媒：韩国国内8英寸半导体代工产能2022年将售罄，涨价不可避免

韩国国内8英寸半导体代工产能的预订实际上已经结束了。预计今年的半导体供应也将因代工产能不足而面临挑战。与去年类似的困境。据ETNews报道，KeyFoundry今年的产能已经被预订满，每六个月接受预订的DBHiTek也将在第二季度关闭订单通道。从三季度的订单情况来看，代工厂的

06-08
Vivo X6速度够快，够流畅，业界将推出全球最强大的Hi-Fi芯片，或者它可能会同时发布耳机，

距离vivo年度旗舰《快到开心》11月30日在北京水立方发布还有几天了，vivo X6也快到让人开心了。这款主打“快”核心功能的机器，近日不仅通过铺天盖地的传播让大家知道了即将发布的新机，今天还联手数十个品牌席卷了整个微博圈。又一个营销奇迹。从功能上来说，vivo X6够快

06-17
阿里X蚂蚁发布“数字木兰”计划，专为女性企业家打造

阿里为女性创业者提供专属金融服务和成长空间，不断推动他们在农村电商、跨境电商、社交经济、新媒体等数字经济领域获得更多创业优势。参与可获得： 1. 每年免息高达20万元 2. 品牌故事及商业经验曝光（万元曝光） 3. 免费课程培训本计划旨在5年内覆盖全球1万元，通过数字化

06-18
芒果旅游获数千万元Pre-A轮融资，文信集团投资

据投资界2月22日消息，芒果旅游近日宣布完成数千万元Pre-A轮融资本轮融资，投资方为文鑫集团。　　据悉，本轮融资完成后，芒果出行将快速增运车辆、拓展城市。同时，融资资金还将用于网站建设、平台技术升级、大数据研发及运营等。　　芒果出行是一家为用户提供分时租赁

06-18