1. RDD到底是什么先把它当成本质是一个数据容器期末复习Spark绕不开的一个核心概念就是RDD。很多同学最开始看官方文档会被弹性分布式数据集这几个字唬住觉得是个很高深的东西。我当年学的时候也卡在这里后来用大白话理解RDD就是一个只读的、分区的、分布式的数据集容器。你可以把它想象成一个超级大的数组但这个数组不是存在单台机器的内存里而是被切成了很多块分区分散在一个集群的多台机器上。它最核心的三个词拆开看分布式数据被切分成多个分区Partition每个分区可以存储在不同的Executor上。集群有多少核、多少内存数据就怎么分散。数据集它本质上是数据的抽象集合可以是内存中的对象也可以是磁盘上的文件还可以是外部存储系统里的数据。弹性这个弹性体现在两个层面一是内存不够的时候可以自动溢写到磁盘二是某个分区丢失后可以根据血统Lineage重新计算出来不会导致整个任务崩溃。RDD的诞生背景其实是为了解决早期MapReduce框架在迭代计算和交互式查询场景下的痛点——频繁落盘导致性能低下。Spark把数据尽量放在内存里做计算速度自然就上去了。对于期末复习来说你不需要把源码都读透但必须能回答这几个问题RDD是个什么东西、它解决了什么问题、它有哪些特性。考试喜欢考概念辨析题比如RDD是可变的数据结构吗——答案是否定的RDD是不可变的每次转换都会生成一个新的RDD原来的RDD保持原样。这个不可变性是容错机制的基础。2. RDD的五大属性和工作原理面试和考试都绕不开的硬核这五点是RDD的骨架也是期末卷子里出现频率极高的考点。我把它们整理成一个方便记忆的表格后面再逐个拆开讲原理。属性名称核心作用一句话记忆分区列表数据被切分成哪些块数据在哪几台机器上分区计算函数每个分区怎么算出数据如何从源头生成数据依赖关系和父RDD之间的血缘关系我爸爸是谁、爷爷是谁分区器数据按什么规则分区哈希还是范围分区首选位置计算时优先调度到哪个节点数据本地性优化2.1 分区列表数据的物理分布每个RDD都维护着一个分区列表每个分区对应数据集的一个逻辑块。比如你在HDFS上读一个文件默认每个Block对应一个分区文件的Block分散在多个DataNode上RDD的分区也就跟着分散在对应的Executor节点上。分区数在创建RDD时可以手动指定也可以由系统自动推断。比如sc.textFile(hdfs://path)默认分片规则受文件块大小和spark.default.parallelism参数影响。考试常考的点是分区数是越大越好吗答案是否定的。分区太多任务调度开销太大分区太少没法充分利用集群的并行能力。一般经验值是每个CPU核心分配2到4个任务。2.2 分区计算函数数据从哪来RDD本身并不存储真正的数据它只记录如何计算出这些数据的函数。这个设计非常巧妙——RDD的每个分区都对应一个迭代器Iterator真正要计算的时候才会去执行这个函数逐条产出数据。这种机制叫做惰性求值Lazy Evaluation也是Spark性能和资源利用的关键。举个形象的例子一个RDD的创建方式如果是sc.parallelize(List(1, 2, 3, 4))那么分区计算函数就是把内存里的这个List按照分区规则切片。如果是sc.textFile(...)计算函数就是读取HDFS对应分片的内容按行解析。重点是理解RDD只在行动算子触发的时候才会真正做计算前面所有的转换操作都只是把计算函数串联起来。2.3 依赖关系容错和DAG调度的根基RDD的依赖分两类窄依赖和宽依赖。窄依赖父RDD的每个分区最多被子RDD的一个分区使用。比如map、filter、union这类依赖不需要洗牌Shuffle计算效率高一个分区出错后只需要重新计算那一个分区。宽依赖父RDD的每个分区可能被子RDD的多个分区使用。比如groupByKey、reduceByKey这类操作一定会产生Shuffle需要把相同Key的数据汇聚到同一个节点上父分区一旦出错可能所有子分区都要重新计算。考试和面试都很喜欢问为什么窄依赖比宽依赖高效因为窄依赖可以在同一个节点上完成流水线式的计算不用跨节点传输数据而且容错恢复的成本更小。我在实际调优的时候会刻意把宽依赖操作尽可能放后面先把数据用窄依赖算子过滤精简这样Shuffle的数据量能少很多。2.4 分区器决定数据怎么分组分区器有三种HashPartitioner、RangePartitioner以及自定义分区器。只有Key-Value类型的RDD才有分区器的概念非KV类型的RDD分区器默认是None。HashPartitioner默认分区器按key.hashCode() % 分区数取模。这种方式的缺点是容易产生数据倾斜尤其当Key分布不均匀时某些分区的数据量会远大于其他分区。RangePartitioner按照Key的区间范围来分区能保证各个分区的数据量大致均衡。它需要先对Key做一次采样确定分区边界。自定义分区器继承Partitioner类重写numPartitions和getPartition方法。实际业务中经常用比如按照地域ID做范围分区尽量把同一地域的数据放在同一个分区减少后续聚合的Shuffle量。2.5 首选位置数据本地性的巧妙设计RDD的每个分区都有一个首选位置Preferred LocationSpark会优先把计算任务调度到数据所在的节点上避免数据在网络间来回传输。这个设计叫数据本地性Data Locality有五个级别PROCESS_LOCAL、NODE_LOCAL、RACK_LOCAL、ANY。级别越高计算效率越高。实际使用中如果任务日志频繁出现Locality level: ANY说明数据本地性没做好任务大部分时间浪费在等待数据拉取上了此时需要考虑调整Executor的内存、分区大小或者存储格式来改善。3. 常用算子分类总结转换算子、行动算子一个都不能少RDD的算子分两大类转换算子Transformation和行动算子Action。这个分类是期末试卷的重点我见过很多同学栽在这个算子到底是转换还是行动的判断上所以先把最核心的判断标准写在这里转换算子是懒执行的返回值还是RDD行动算子会立即触发真正的计算返回值不再是RDD可能是数组、数值、字典等。3.1 高频转换算子map、flatMap、filter的细节差异这几个算子是最基础的但细节上容易出题。map一对一映射输入一条数据输出一条数据。比如rdd.map(x x * 2)每条数据都会被处理一次返回的新RDD元素个数和原来一样。flatMap一对多映射先把每条数据处理成一个集合再把集合里的元素摊平成一个新的RDD。比如你有一行文本hello world想把它拆成单词列表就必须用flatMap(_.split( ))如果用map得到的是RDD[Array[String]]而不是RDD[String]。filter按条件过滤保留返回true的元素。注意条件函数要用判断表达式别写成赋值表达式这个低级错误每年都有人犯。此外还有distinct去重、union合并两个RDD、intersection取交集、subtract取差集、sample采样。前后两者之间union要求两个RDD的类型一致而且它不会去重两个RDD里重复的元素会同时保留下来。map和flatMap的实际对比我给你们写一段演示性质的伪代码输入数据RDD(hello spark, hello scala) map(_.split( )) 的结果RDD(Array(hello,spark), Array(hello,scala)) flatMap(_.split( )) 的结果RDD(hello,spark,hello,scala)这个例子几乎每年都会出现在选择题或者填空题里务必理解透彻。3.2 键值型算子reduceByKey、groupByKey、aggregateByKey键值型RDDPairRDD的算子是Spark的核心战场考试的大题基本从这里出。reduceByKey(func)按照Key分组再用给定的函数对每个分组内的Value做归约。它会在Map端先做一次预聚合Combine减少Shuffle数据量效率高日常开发首选。groupByKey()按照Key分组但不做预聚合把每个Key对应的所有Value原封不动地收集到同一个列表里。Shuffle数据量巨大性能差能不用尽量不用。aggregateByKey(zeroValue)(seqOp, combOp)可以同时对每个分区内的数据做局部聚合再跨分区做全局聚合灵活度最高。它有三个参数初值、分区内聚合函数、分区间聚合函数。注意zeroValue会在每个分区内各参与一次计算不是只参与一次。mapValues只对Value做映射Key保持不变。这是个窄依赖算子不会触发Shuffle。实际写代码时能先mapValues就别急着map。keys / values分别取出键集合和值集合。我个人的实践体会是能用reduceByKey绝不用groupByKey。在几个G的数据集上两者性能差距肉眼可见。如果抽象任务明确要求先分组再排序可以用sortByKey配合groupByKey但也要权衡数据量是否可控。3.3 行动算子什么时候真正触发计算行动算子是Spark DAG执行的开关。常见的Action包括collect()把RDD的所有元素拉回Driver端以数组形式返回。数据量大的时候严禁直接collect会把Driver内存打爆。count()统计元素个数。first()取第一个元素。take(n)取前n个元素。reduce(func)对RDD里的所有元素做归约。注意它和reduceByKey的区别——reduce是行动算子reduceByKey是转换算子。foreach(func)对每个元素执行一个函数常用于输出到外部存储。saveAsTextFile(path)把RDD保存成文本文件。考试常考的知识点是一个Action对应一个Job。比如你在代码里调用两次count()就会产生两个Job分别走一遍DAG调度。这个设定对理解Spark的执行模型很重要也是期末卷子里的送分题。3.4 持久化算子cache、persist、checkpoint这仨不是转换也不是行动但对性能影响极大期末复习不要漏掉。cache()等价于persist(StorageLevel.MEMORY_ONLY)把RDD的计算结果缓存在内存里。persist(StorageLevel)可以自定义存储级别比如MEMORY_AND_DISK、DISK_ONLY、MEMORY_ONLY_2副本数2。checkpoint()把RDD的数据快照写到可靠的存储比如HDFS上同时切断血缘关系。它们的区别在于cache和persist不会改变血缘关系如果数据丢失了还能通过血统重新计算checkpoint会切断之前的血缘依赖适合那些计算链路特别长、中间结果特别重要的场景。我个人的调优习惯是如果一个RDD会被多个Action反复用到先用cache把它留住如果这个RDD的血统链很长且数据量又大考虑checkpoint。4. 经典的RDD练习题拿来就能刷刷完就能上考场这部分我从历年期末题和实际工作里常用的场景里挑选了几个有代表性的练习题代码我会用Scala写因为Spark原生接口就是Scala期末考的也基本是Scala语法。每道题都会给出解题思路和参考实现建议你先自己写一遍再看答案效果完全不一样。4.1 统计单词个数大数据入门的第一课需求给定一个文本文件路径统计每个单词出现的次数结果按次数降序排列输出前10个单词。解题思路flatMap切词map生成键值对reduceByKey汇总sortBy排序take取前10个。参考代码Scalaval rdd sc.textFile(hdfs:///user/data/words.txt) val result rdd .flatMap(_.split( )) .map(word (word, 1)) .reduceByKey(_ _) .sortBy(_._2, ascending false) .take(10) result.foreach(println)细节提醒sortBy(_._2, ascending false)里的_._2表示元组的第二个字段也就是词频。如果不指定ascending默认是升序这里必须写false才能降序。这道题背后的知识点很密集flatMap的作用、map构造KV、reduceByKey的预聚合、sortBy和take的配合。把这道题的每个步骤都搞明白RDD的基础就算过关了。4.2 分组求TopN考察宽依赖和排序的综合题需求有一个销售数据RDD格式是(商品分类, 销售额)要求统计每个分类下销售额最高的3个商品输出格式为(分类, 销售额列表Top3)。解题思路先groupByKey拿到每个分类下的所有销售记录再对每个分组内部的迭代器做排序取前3。参考代码val data sc.parallelize(Seq( (手机, 100), (手机, 200), (手机, 50), (电脑, 300), (电脑, 150), (电脑, 80), (电脑, 500) )) val result data .groupByKey() .mapValues(values values.toList.sorted(Ordering[Int].reverse).take(3)) result.collect().foreach(println)如果数据集特别大不建议用groupByKey单打独斗可以先用repartitionAndSortWithinPartitions配合自定义分区器让每个分区内数据有序再在每个分区内取TopN。这个方法在面试里讲出来会很加分但期末如果时间紧张groupByKey版本就足够了。4.3 二次排序面试笔试都爱出的硬骨头需求一个RDD的元素是(学科, 姓名, 分数)请按照学科分组组内按照分数降序、姓名升序排列。解题思路自定义一个排序类继承Ordered或者实现Ordering接口作为sortBy的排序规则。参考代码片段case class ScoreRecord(subject: String, name: String, score: Int) implicit val ordering: Ordering[ScoreRecord] Ordering.by(r (r.subject, -r.score, r.name)) val data sc.parallelize(Seq( ScoreRecord(数学, 张三, 90), ScoreRecord(数学, 李四, 85), ScoreRecord(语文, 王五, 92) )) val result data.sortBy(identity) result.collect().foreach(println)这里用-r.score实现降序因为Ordering默认是升序。如果你用Scala的sortBy处理Tuple规则会更加直接sortBy(r (r._1, -r._2))。二次排序的关键不是代码量而是理解排序规则的组合方式。如果题目要求多字段排序就构造一个组合Key注意每个字段的升降序方向。4.4 数据去重与计数考察集合算子需求有两个RDD分别包含用户ID一个是登录用户表一个是购买用户表统计同时在两张表里出现的用户数以及至少在一张表里出现的用户数。解题思路intersection求两者都有的union后再distinct求总用户数。参考代码val loginUsers sc.parallelize(Seq(1, 2, 3, 4, 5)) val buyUsers sc.parallelize(Seq(4, 5, 6, 7)) val bothCount loginUsers.intersection(buyUsers).count() val totalCount loginUsers.union(buyUsers).distinct().count() println(s同时出现: $bothCount, 至少出现: $totalCount)很多人容易在union之后忘记distinct导致用户数翻倍。这是个经典的坑考试的时候宁可多写一步distinct也别让数算错。这里再补充一个去重的高频细节distinct的底层实现其实是reduceByKey把value统一置空再取Key所以它也是一个会产生Shuffle的操作。数据量特别大的时候distinct比较耗时可以先做一轮map((_, null))再reduceByKey效果基本一样。4.5 求平均值考察aggregate的灵活使用需求一个RDD包含一组整数求它们的平均值。要求分别使用reduce和aggregate两种方式实现。参考代码val nums sc.parallelize(Seq(1, 2, 3, 4, 5, 6)) // 方式一reduce先求和再除以个数 val sum1 nums.reduce(_ _) val avg1 sum1.toDouble / nums.count() // 方式二aggregate一次计算实现 val (sum2, count2) nums.aggregate((0, 0))( (acc, num) (acc._1 num, acc._2 1), (acc1, acc2) (acc1._1 acc2._1, acc1._2 acc2._2) ) val avg2 sum2.toDouble / count2 println(savg1$avg1, avg2$avg2)aggregate的写法是重点。第一个括号里的(0, 0)是初始值表示总和为0、计数为0第二个括号里第一个函数负责分区内累加第二个函数负责分区间合并。注意两个函数都要把总和和计数同时处理缺一个都会算错。这道题的考察点实际上是大家对Action类型算子的理解程度。reduce和aggregate都不会返回RDD所以它们是行动算子会直接触发Job计算。5. 期末复习避坑指南这些坑我替你们踩过了5.1 概念辨析题的高频陷阱RDD这块的期末考题很喜欢出辨析题下面的表格是我总结的常考陷阱建议考前过一遍错误说法正确理解RDD存储了真实的数据RDD只存储元数据和计算函数不存储真实数据RDD创建后可以修改元素RDD不可变每次转换都生成新RDDcache会切断血统关系cache不会切断血缘checkpoint才会groupByKey比reduceByKey效率高两者都会Shuffle但reduceByKey有Map端预聚合更高效窄依赖一定会产生Shuffle窄依赖不产生Shuffle宽依赖才产生RDD必须有分区器只有KV类型RDD才有分区器普通RDD分区器为None这些都是我自己和身边同学当初复习时反复弄错的地方尤其RDD到底存不存数据这一点几乎每学期都有人在这丢分。5.2 实战代码的常见报错与排查记录平时刷题或者做课程设计的时候最常遇到的三个报错我分别说下原因和解决办法。第一个报错org.apache.spark.SparkException: Job aborted due to stage failure这个报错太常见了绝大部分时候是数据倾斜导致的。排查方法看Spark UI里的Stage页面如果某个Task处理的数据量远超其他Task基本可以确定是倾斜。解决办法对Key做加盐salting处理也就是给Key加上随机前缀把大Key拆散到多个分区。调整分区数增大并行度。如果倾斜不严重可以考虑用reduceByKey代替groupByKey因为预聚合能先过滤一部分数据。第二个报错java.lang.OutOfMemoryError: Java heap spaceExecutor内存不够。原因通常是先collect()了太大的RDD或者缓存了过多数据又或者并行度过高导致每个Task分摊的堆内存不足。解决办法去掉collect()改用saveAsTextFile写入外部存储。调大spark.executor.memory但别超过单个节点的物理内存。检查是否滥用cache()用完记得unpersist()。第三个报错org.apache.spark.shuffle.FetchFailedExceptionShuffle拉取失败底层常见原因是某个Executor宕机或者网络抖动。解决办法先确认节点健康状态再把spark.shuffle.io.maxRetries调大到5以上、spark.shuffle.io.retryWait调到10秒以上给重试留出足够窗口。5.3 代码只能写到第X步运行时永久卡住怎么排查有一种同学经常遇到的情况是小数据集跑得好好的换到几GB的数据就卡死。这种问题几乎都是因为某个宽依赖算子产生的数据量超出预期。我之前做一个课程设计用groupByKey处理一堆分类标签结果某个分类的Key下面挂了上亿条记录单机任务直接卡死。排查套路我整理成了几个步骤先看Spark UI找到运行时间最长的Stage确认是不是某个Action触发的。点进那个Stage看每个Task的处理时间和Shuffle读写量判断是否存在数据倾斜。如果存在倾斜把倾斜Key的明细打印出来看一下它的数据量占比。对症下药倾斜Key不多就用过滤或者加盐整体数据不均匀就换RangePartitioner如果只是某个特定计算太重考虑改写算法逻辑。卡死的问题十次里有八次是数据倾斜剩下两次是Spark配置和物理资源不匹配。排查思路按上面这个顺序走基本不会白折腾。5.4 考试答题的小技巧怎么答才能拿高分最后分享几个我总结的答题技巧。这部分可能比刷题本身更实用因为期末卷子很多题目考的是表达和逻辑。第一个技巧凡是问RDD特性从五大属性入手答。分点列出分区列表、计算函数、依赖关系、分区器、首选位置再展开说每条对应什么作用基本就是满分结构。第二个技巧凡是问为什么某个算子比另一个快一定要说到Shuffle和预聚合。比如问为什么reduceByKey比groupByKey快你只要把这个逻辑说清楚阅卷老师就知道你真正理解了。第三个技巧凡是写代码题先写清楚流程再写具体代码。阅卷的时候代码对了一部分也能拿到步骤分但如果只写个最终答案没有过程错了就直接归零。我当年考试就是习惯先把flatMap切分——map生成KV——reduceByKey汇总——sortBy排序——take取前N这串流程图写出来再逐行实现得分稳定很多。第四个技巧涉及分区数计算的题别忘公式。比如sc.parallelize默认分区数的计算逻辑textFile默认最小分区数这些细节公式要背熟。不用死记源码但spark.default.parallelism的值怎么算出来的各个模式下的默认值是多少最好能随口说出来。6. 一个综合案例串起所有知识点理论说太多容易散我再补一个综合性的练兵案例把前面讲到的知识点全部串起来。这个案例我自己在课程设计里做过一个简化版很适合期末前用来自查。场景给定一份电商订单日志RDD每行格式为订单ID, 用户ID, 商品分类, 商品金额。要求完成以下几个统计任务每个商品分类的总销售额。每个商品分类下销售额最高的前3个订单。消费总额排名前5的用户ID。各个分类的订单数量占比。参考实现思路是// 假设原始格式是字符串先用map解析成元组 val orders sc.textFile(hdfs:///data/orders.log).map { line val arr line.split(,) (arr(0), arr(1), arr(2), arr(3).toDouble) } // 任务1分类总销售额 val categoryTotal orders .map(order (order._3, order._4)) .reduceByKey(_ _) // 任务2每个分类下订单金额最高的3笔 val categoryTop3 orders .map(order (order._3, (order._1, order._4))) .groupByKey() .mapValues(it it.toList.sortBy(-_._2).take(3)) // 任务3消费总额排名前5的用户先按用户累加再倒序取前5 val topUsers orders .map(order (order._2, order._4)) .reduceByKey(_ _) .sortBy(_._2, ascending false) .take(5) // 任务4各分类订单数量占比 val totalOrders orders.count() val categoryCount orders .map(order (order._3, 1L)) .reduceByKey(_ _) .mapValues(count count.toDouble / totalOrders)做完这个案例你对RDD的map、flatMap、reduceByKey、groupByKey、mapValues、sortBy、take、count这些高频算子都会有整体感知。剩下的就是重复练习直到不看答案也能顺手写出来。我在实际操作中的一点感受是RDD部分的期末复习真的不是靠死记硬背而是要亲手把代码敲一遍看一遍Spark UI上的执行流程再回头对照概念。数据倾斜、宽窄依赖、惰性计算这些知识点只要你亲手跑过几个大一点的作业就会理解得特别透彻。最后再提醒一句考试前把reduceByKey和groupByKey的区别、RDD的五大属性、转换算子和行动算子的判断标准这三块看熟基本就能覆盖八成以上的分数。
1. RDD到底是什么先把它当成本质是一个数据容器期末复习Spark绕不开的一个核心概念就是RDD。很多同学最开始看官方文档会被弹性分布式数据集这几个字唬住觉得是个很高深的东西。我当年学的时候也卡在这里后来用大白话理解RDD就是一个只读的、分区的、分布式的数据集容器。你可以把它想象成一个超级大的数组但这个数组不是存在单台机器的内存里而是被切成了很多块分区分散在一个集群的多台机器上。它最核心的三个词拆开看分布式数据被切分成多个分区Partition每个分区可以存储在不同的Executor上。集群有多少核、多少内存数据就怎么分散。数据集它本质上是数据的抽象集合可以是内存中的对象也可以是磁盘上的文件还可以是外部存储系统里的数据。弹性这个弹性体现在两个层面一是内存不够的时候可以自动溢写到磁盘二是某个分区丢失后可以根据血统Lineage重新计算出来不会导致整个任务崩溃。RDD的诞生背景其实是为了解决早期MapReduce框架在迭代计算和交互式查询场景下的痛点——频繁落盘导致性能低下。Spark把数据尽量放在内存里做计算速度自然就上去了。对于期末复习来说你不需要把源码都读透但必须能回答这几个问题RDD是个什么东西、它解决了什么问题、它有哪些特性。考试喜欢考概念辨析题比如RDD是可变的数据结构吗——答案是否定的RDD是不可变的每次转换都会生成一个新的RDD原来的RDD保持原样。这个不可变性是容错机制的基础。2. RDD的五大属性和工作原理面试和考试都绕不开的硬核这五点是RDD的骨架也是期末卷子里出现频率极高的考点。我把它们整理成一个方便记忆的表格后面再逐个拆开讲原理。属性名称核心作用一句话记忆分区列表数据被切分成哪些块数据在哪几台机器上分区计算函数每个分区怎么算出数据如何从源头生成数据依赖关系和父RDD之间的血缘关系我爸爸是谁、爷爷是谁分区器数据按什么规则分区哈希还是范围分区首选位置计算时优先调度到哪个节点数据本地性优化2.1 分区列表数据的物理分布每个RDD都维护着一个分区列表每个分区对应数据集的一个逻辑块。比如你在HDFS上读一个文件默认每个Block对应一个分区文件的Block分散在多个DataNode上RDD的分区也就跟着分散在对应的Executor节点上。分区数在创建RDD时可以手动指定也可以由系统自动推断。比如sc.textFile(hdfs://path)默认分片规则受文件块大小和spark.default.parallelism参数影响。考试常考的点是分区数是越大越好吗答案是否定的。分区太多任务调度开销太大分区太少没法充分利用集群的并行能力。一般经验值是每个CPU核心分配2到4个任务。2.2 分区计算函数数据从哪来RDD本身并不存储真正的数据它只记录如何计算出这些数据的函数。这个设计非常巧妙——RDD的每个分区都对应一个迭代器Iterator真正要计算的时候才会去执行这个函数逐条产出数据。这种机制叫做惰性求值Lazy Evaluation也是Spark性能和资源利用的关键。举个形象的例子一个RDD的创建方式如果是sc.parallelize(List(1, 2, 3, 4))那么分区计算函数就是把内存里的这个List按照分区规则切片。如果是sc.textFile(...)计算函数就是读取HDFS对应分片的内容按行解析。重点是理解RDD只在行动算子触发的时候才会真正做计算前面所有的转换操作都只是把计算函数串联起来。2.3 依赖关系容错和DAG调度的根基RDD的依赖分两类窄依赖和宽依赖。窄依赖父RDD的每个分区最多被子RDD的一个分区使用。比如map、filter、union这类依赖不需要洗牌Shuffle计算效率高一个分区出错后只需要重新计算那一个分区。宽依赖父RDD的每个分区可能被子RDD的多个分区使用。比如groupByKey、reduceByKey这类操作一定会产生Shuffle需要把相同Key的数据汇聚到同一个节点上父分区一旦出错可能所有子分区都要重新计算。考试和面试都很喜欢问为什么窄依赖比宽依赖高效因为窄依赖可以在同一个节点上完成流水线式的计算不用跨节点传输数据而且容错恢复的成本更小。我在实际调优的时候会刻意把宽依赖操作尽可能放后面先把数据用窄依赖算子过滤精简这样Shuffle的数据量能少很多。2.4 分区器决定数据怎么分组分区器有三种HashPartitioner、RangePartitioner以及自定义分区器。只有Key-Value类型的RDD才有分区器的概念非KV类型的RDD分区器默认是None。HashPartitioner默认分区器按key.hashCode() % 分区数取模。这种方式的缺点是容易产生数据倾斜尤其当Key分布不均匀时某些分区的数据量会远大于其他分区。RangePartitioner按照Key的区间范围来分区能保证各个分区的数据量大致均衡。它需要先对Key做一次采样确定分区边界。自定义分区器继承Partitioner类重写numPartitions和getPartition方法。实际业务中经常用比如按照地域ID做范围分区尽量把同一地域的数据放在同一个分区减少后续聚合的Shuffle量。2.5 首选位置数据本地性的巧妙设计RDD的每个分区都有一个首选位置Preferred LocationSpark会优先把计算任务调度到数据所在的节点上避免数据在网络间来回传输。这个设计叫数据本地性Data Locality有五个级别PROCESS_LOCAL、NODE_LOCAL、RACK_LOCAL、ANY。级别越高计算效率越高。实际使用中如果任务日志频繁出现Locality level: ANY说明数据本地性没做好任务大部分时间浪费在等待数据拉取上了此时需要考虑调整Executor的内存、分区大小或者存储格式来改善。3. 常用算子分类总结转换算子、行动算子一个都不能少RDD的算子分两大类转换算子Transformation和行动算子Action。这个分类是期末试卷的重点我见过很多同学栽在这个算子到底是转换还是行动的判断上所以先把最核心的判断标准写在这里转换算子是懒执行的返回值还是RDD行动算子会立即触发真正的计算返回值不再是RDD可能是数组、数值、字典等。3.1 高频转换算子map、flatMap、filter的细节差异这几个算子是最基础的但细节上容易出题。map一对一映射输入一条数据输出一条数据。比如rdd.map(x x * 2)每条数据都会被处理一次返回的新RDD元素个数和原来一样。flatMap一对多映射先把每条数据处理成一个集合再把集合里的元素摊平成一个新的RDD。比如你有一行文本hello world想把它拆成单词列表就必须用flatMap(_.split( ))如果用map得到的是RDD[Array[String]]而不是RDD[String]。filter按条件过滤保留返回true的元素。注意条件函数要用判断表达式别写成赋值表达式这个低级错误每年都有人犯。此外还有distinct去重、union合并两个RDD、intersection取交集、subtract取差集、sample采样。前后两者之间union要求两个RDD的类型一致而且它不会去重两个RDD里重复的元素会同时保留下来。map和flatMap的实际对比我给你们写一段演示性质的伪代码输入数据RDD(hello spark, hello scala) map(_.split( )) 的结果RDD(Array(hello,spark), Array(hello,scala)) flatMap(_.split( )) 的结果RDD(hello,spark,hello,scala)这个例子几乎每年都会出现在选择题或者填空题里务必理解透彻。3.2 键值型算子reduceByKey、groupByKey、aggregateByKey键值型RDDPairRDD的算子是Spark的核心战场考试的大题基本从这里出。reduceByKey(func)按照Key分组再用给定的函数对每个分组内的Value做归约。它会在Map端先做一次预聚合Combine减少Shuffle数据量效率高日常开发首选。groupByKey()按照Key分组但不做预聚合把每个Key对应的所有Value原封不动地收集到同一个列表里。Shuffle数据量巨大性能差能不用尽量不用。aggregateByKey(zeroValue)(seqOp, combOp)可以同时对每个分区内的数据做局部聚合再跨分区做全局聚合灵活度最高。它有三个参数初值、分区内聚合函数、分区间聚合函数。注意zeroValue会在每个分区内各参与一次计算不是只参与一次。mapValues只对Value做映射Key保持不变。这是个窄依赖算子不会触发Shuffle。实际写代码时能先mapValues就别急着map。keys / values分别取出键集合和值集合。我个人的实践体会是能用reduceByKey绝不用groupByKey。在几个G的数据集上两者性能差距肉眼可见。如果抽象任务明确要求先分组再排序可以用sortByKey配合groupByKey但也要权衡数据量是否可控。3.3 行动算子什么时候真正触发计算行动算子是Spark DAG执行的开关。常见的Action包括collect()把RDD的所有元素拉回Driver端以数组形式返回。数据量大的时候严禁直接collect会把Driver内存打爆。count()统计元素个数。first()取第一个元素。take(n)取前n个元素。reduce(func)对RDD里的所有元素做归约。注意它和reduceByKey的区别——reduce是行动算子reduceByKey是转换算子。foreach(func)对每个元素执行一个函数常用于输出到外部存储。saveAsTextFile(path)把RDD保存成文本文件。考试常考的知识点是一个Action对应一个Job。比如你在代码里调用两次count()就会产生两个Job分别走一遍DAG调度。这个设定对理解Spark的执行模型很重要也是期末卷子里的送分题。3.4 持久化算子cache、persist、checkpoint这仨不是转换也不是行动但对性能影响极大期末复习不要漏掉。cache()等价于persist(StorageLevel.MEMORY_ONLY)把RDD的计算结果缓存在内存里。persist(StorageLevel)可以自定义存储级别比如MEMORY_AND_DISK、DISK_ONLY、MEMORY_ONLY_2副本数2。checkpoint()把RDD的数据快照写到可靠的存储比如HDFS上同时切断血缘关系。它们的区别在于cache和persist不会改变血缘关系如果数据丢失了还能通过血统重新计算checkpoint会切断之前的血缘依赖适合那些计算链路特别长、中间结果特别重要的场景。我个人的调优习惯是如果一个RDD会被多个Action反复用到先用cache把它留住如果这个RDD的血统链很长且数据量又大考虑checkpoint。4. 经典的RDD练习题拿来就能刷刷完就能上考场这部分我从历年期末题和实际工作里常用的场景里挑选了几个有代表性的练习题代码我会用Scala写因为Spark原生接口就是Scala期末考的也基本是Scala语法。每道题都会给出解题思路和参考实现建议你先自己写一遍再看答案效果完全不一样。4.1 统计单词个数大数据入门的第一课需求给定一个文本文件路径统计每个单词出现的次数结果按次数降序排列输出前10个单词。解题思路flatMap切词map生成键值对reduceByKey汇总sortBy排序take取前10个。参考代码Scalaval rdd sc.textFile(hdfs:///user/data/words.txt) val result rdd .flatMap(_.split( )) .map(word (word, 1)) .reduceByKey(_ _) .sortBy(_._2, ascending false) .take(10) result.foreach(println)细节提醒sortBy(_._2, ascending false)里的_._2表示元组的第二个字段也就是词频。如果不指定ascending默认是升序这里必须写false才能降序。这道题背后的知识点很密集flatMap的作用、map构造KV、reduceByKey的预聚合、sortBy和take的配合。把这道题的每个步骤都搞明白RDD的基础就算过关了。4.2 分组求TopN考察宽依赖和排序的综合题需求有一个销售数据RDD格式是(商品分类, 销售额)要求统计每个分类下销售额最高的3个商品输出格式为(分类, 销售额列表Top3)。解题思路先groupByKey拿到每个分类下的所有销售记录再对每个分组内部的迭代器做排序取前3。参考代码val data sc.parallelize(Seq( (手机, 100), (手机, 200), (手机, 50), (电脑, 300), (电脑, 150), (电脑, 80), (电脑, 500) )) val result data .groupByKey() .mapValues(values values.toList.sorted(Ordering[Int].reverse).take(3)) result.collect().foreach(println)如果数据集特别大不建议用groupByKey单打独斗可以先用repartitionAndSortWithinPartitions配合自定义分区器让每个分区内数据有序再在每个分区内取TopN。这个方法在面试里讲出来会很加分但期末如果时间紧张groupByKey版本就足够了。4.3 二次排序面试笔试都爱出的硬骨头需求一个RDD的元素是(学科, 姓名, 分数)请按照学科分组组内按照分数降序、姓名升序排列。解题思路自定义一个排序类继承Ordered或者实现Ordering接口作为sortBy的排序规则。参考代码片段case class ScoreRecord(subject: String, name: String, score: Int) implicit val ordering: Ordering[ScoreRecord] Ordering.by(r (r.subject, -r.score, r.name)) val data sc.parallelize(Seq( ScoreRecord(数学, 张三, 90), ScoreRecord(数学, 李四, 85), ScoreRecord(语文, 王五, 92) )) val result data.sortBy(identity) result.collect().foreach(println)这里用-r.score实现降序因为Ordering默认是升序。如果你用Scala的sortBy处理Tuple规则会更加直接sortBy(r (r._1, -r._2))。二次排序的关键不是代码量而是理解排序规则的组合方式。如果题目要求多字段排序就构造一个组合Key注意每个字段的升降序方向。4.4 数据去重与计数考察集合算子需求有两个RDD分别包含用户ID一个是登录用户表一个是购买用户表统计同时在两张表里出现的用户数以及至少在一张表里出现的用户数。解题思路intersection求两者都有的union后再distinct求总用户数。参考代码val loginUsers sc.parallelize(Seq(1, 2, 3, 4, 5)) val buyUsers sc.parallelize(Seq(4, 5, 6, 7)) val bothCount loginUsers.intersection(buyUsers).count() val totalCount loginUsers.union(buyUsers).distinct().count() println(s同时出现: $bothCount, 至少出现: $totalCount)很多人容易在union之后忘记distinct导致用户数翻倍。这是个经典的坑考试的时候宁可多写一步distinct也别让数算错。这里再补充一个去重的高频细节distinct的底层实现其实是reduceByKey把value统一置空再取Key所以它也是一个会产生Shuffle的操作。数据量特别大的时候distinct比较耗时可以先做一轮map((_, null))再reduceByKey效果基本一样。4.5 求平均值考察aggregate的灵活使用需求一个RDD包含一组整数求它们的平均值。要求分别使用reduce和aggregate两种方式实现。参考代码val nums sc.parallelize(Seq(1, 2, 3, 4, 5, 6)) // 方式一reduce先求和再除以个数 val sum1 nums.reduce(_ _) val avg1 sum1.toDouble / nums.count() // 方式二aggregate一次计算实现 val (sum2, count2) nums.aggregate((0, 0))( (acc, num) (acc._1 num, acc._2 1), (acc1, acc2) (acc1._1 acc2._1, acc1._2 acc2._2) ) val avg2 sum2.toDouble / count2 println(savg1$avg1, avg2$avg2)aggregate的写法是重点。第一个括号里的(0, 0)是初始值表示总和为0、计数为0第二个括号里第一个函数负责分区内累加第二个函数负责分区间合并。注意两个函数都要把总和和计数同时处理缺一个都会算错。这道题的考察点实际上是大家对Action类型算子的理解程度。reduce和aggregate都不会返回RDD所以它们是行动算子会直接触发Job计算。5. 期末复习避坑指南这些坑我替你们踩过了5.1 概念辨析题的高频陷阱RDD这块的期末考题很喜欢出辨析题下面的表格是我总结的常考陷阱建议考前过一遍错误说法正确理解RDD存储了真实的数据RDD只存储元数据和计算函数不存储真实数据RDD创建后可以修改元素RDD不可变每次转换都生成新RDDcache会切断血统关系cache不会切断血缘checkpoint才会groupByKey比reduceByKey效率高两者都会Shuffle但reduceByKey有Map端预聚合更高效窄依赖一定会产生Shuffle窄依赖不产生Shuffle宽依赖才产生RDD必须有分区器只有KV类型RDD才有分区器普通RDD分区器为None这些都是我自己和身边同学当初复习时反复弄错的地方尤其RDD到底存不存数据这一点几乎每学期都有人在这丢分。5.2 实战代码的常见报错与排查记录平时刷题或者做课程设计的时候最常遇到的三个报错我分别说下原因和解决办法。第一个报错org.apache.spark.SparkException: Job aborted due to stage failure这个报错太常见了绝大部分时候是数据倾斜导致的。排查方法看Spark UI里的Stage页面如果某个Task处理的数据量远超其他Task基本可以确定是倾斜。解决办法对Key做加盐salting处理也就是给Key加上随机前缀把大Key拆散到多个分区。调整分区数增大并行度。如果倾斜不严重可以考虑用reduceByKey代替groupByKey因为预聚合能先过滤一部分数据。第二个报错java.lang.OutOfMemoryError: Java heap spaceExecutor内存不够。原因通常是先collect()了太大的RDD或者缓存了过多数据又或者并行度过高导致每个Task分摊的堆内存不足。解决办法去掉collect()改用saveAsTextFile写入外部存储。调大spark.executor.memory但别超过单个节点的物理内存。检查是否滥用cache()用完记得unpersist()。第三个报错org.apache.spark.shuffle.FetchFailedExceptionShuffle拉取失败底层常见原因是某个Executor宕机或者网络抖动。解决办法先确认节点健康状态再把spark.shuffle.io.maxRetries调大到5以上、spark.shuffle.io.retryWait调到10秒以上给重试留出足够窗口。5.3 代码只能写到第X步运行时永久卡住怎么排查有一种同学经常遇到的情况是小数据集跑得好好的换到几GB的数据就卡死。这种问题几乎都是因为某个宽依赖算子产生的数据量超出预期。我之前做一个课程设计用groupByKey处理一堆分类标签结果某个分类的Key下面挂了上亿条记录单机任务直接卡死。排查套路我整理成了几个步骤先看Spark UI找到运行时间最长的Stage确认是不是某个Action触发的。点进那个Stage看每个Task的处理时间和Shuffle读写量判断是否存在数据倾斜。如果存在倾斜把倾斜Key的明细打印出来看一下它的数据量占比。对症下药倾斜Key不多就用过滤或者加盐整体数据不均匀就换RangePartitioner如果只是某个特定计算太重考虑改写算法逻辑。卡死的问题十次里有八次是数据倾斜剩下两次是Spark配置和物理资源不匹配。排查思路按上面这个顺序走基本不会白折腾。5.4 考试答题的小技巧怎么答才能拿高分最后分享几个我总结的答题技巧。这部分可能比刷题本身更实用因为期末卷子很多题目考的是表达和逻辑。第一个技巧凡是问RDD特性从五大属性入手答。分点列出分区列表、计算函数、依赖关系、分区器、首选位置再展开说每条对应什么作用基本就是满分结构。第二个技巧凡是问为什么某个算子比另一个快一定要说到Shuffle和预聚合。比如问为什么reduceByKey比groupByKey快你只要把这个逻辑说清楚阅卷老师就知道你真正理解了。第三个技巧凡是写代码题先写清楚流程再写具体代码。阅卷的时候代码对了一部分也能拿到步骤分但如果只写个最终答案没有过程错了就直接归零。我当年考试就是习惯先把flatMap切分——map生成KV——reduceByKey汇总——sortBy排序——take取前N这串流程图写出来再逐行实现得分稳定很多。第四个技巧涉及分区数计算的题别忘公式。比如sc.parallelize默认分区数的计算逻辑textFile默认最小分区数这些细节公式要背熟。不用死记源码但spark.default.parallelism的值怎么算出来的各个模式下的默认值是多少最好能随口说出来。6. 一个综合案例串起所有知识点理论说太多容易散我再补一个综合性的练兵案例把前面讲到的知识点全部串起来。这个案例我自己在课程设计里做过一个简化版很适合期末前用来自查。场景给定一份电商订单日志RDD每行格式为订单ID, 用户ID, 商品分类, 商品金额。要求完成以下几个统计任务每个商品分类的总销售额。每个商品分类下销售额最高的前3个订单。消费总额排名前5的用户ID。各个分类的订单数量占比。参考实现思路是// 假设原始格式是字符串先用map解析成元组 val orders sc.textFile(hdfs:///data/orders.log).map { line val arr line.split(,) (arr(0), arr(1), arr(2), arr(3).toDouble) } // 任务1分类总销售额 val categoryTotal orders .map(order (order._3, order._4)) .reduceByKey(_ _) // 任务2每个分类下订单金额最高的3笔 val categoryTop3 orders .map(order (order._3, (order._1, order._4))) .groupByKey() .mapValues(it it.toList.sortBy(-_._2).take(3)) // 任务3消费总额排名前5的用户先按用户累加再倒序取前5 val topUsers orders .map(order (order._2, order._4)) .reduceByKey(_ _) .sortBy(_._2, ascending false) .take(5) // 任务4各分类订单数量占比 val totalOrders orders.count() val categoryCount orders .map(order (order._3, 1L)) .reduceByKey(_ _) .mapValues(count count.toDouble / totalOrders)做完这个案例你对RDD的map、flatMap、reduceByKey、groupByKey、mapValues、sortBy、take、count这些高频算子都会有整体感知。剩下的就是重复练习直到不看答案也能顺手写出来。我在实际操作中的一点感受是RDD部分的期末复习真的不是靠死记硬背而是要亲手把代码敲一遍看一遍Spark UI上的执行流程再回头对照概念。数据倾斜、宽窄依赖、惰性计算这些知识点只要你亲手跑过几个大一点的作业就会理解得特别透彻。最后再提醒一句考试前把reduceByKey和groupByKey的区别、RDD的五大属性、转换算子和行动算子的判断标准这三块看熟基本就能覆盖八成以上的分数。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
