1. 项目概述1.1 核心需求解析先说句实在话但凡做过两三个Spark项目的人基本都绕不开词频统计这个入门必经环节。很多教程把它当成“大数据领域的Hello World”这个定位既准确又容易误导人——因为轻视它的人会在后续真正处理业务数据时吃亏。我见过太多简历上写着“熟悉Spark”的候选人一聊到词频统计背后的shuffle机制、数据倾斜问题就支支吾吾说不清楚。词频统计这个项目本质上要做的事情可以用一句话概括给定一批文本文件统计每个单词在整个数据集中出现的总次数。听起来简单但它的技术纵深一点也不浅。你既要考虑单机模式下Python几行代码就能搞定的事情为什么要放到分布式环境里去跑又要思考当数据量从MB级膨胀到TB级时内存、网络、磁盘的瓶颈会出现在哪里还要面对现实世界的数据远比教程里的示例脏得多标点符号、大小写、编码格式、甚至乱码都会让你的“简单需求”变得复杂。从需求拆解的角度来说这个项目至少要覆盖五个层次输入数据的读取与解析、文本内容的切分与清洗、单词的聚合与计数、结果的排序与输出、以及最容易被忽略的——分布式执行过程中的容错与调优。每一层都有值得展开细说的门道。1.2 项目价值与适用场景这个项目适合谁来跑我个人的判断是三类人第一类是刚接触分布式计算的初学者需要通过一个完整的实践闭环建立对Spark编程模型的直觉第二类是已经会写Spark代码但只停留在“能跑就行”阶段的开发者这篇文章里很多调优细节和踩坑经验能帮他们把代码从“跑得通”推进到“跑得快”第三类是准备面试的求职者词频统计往往会被拿来考察Spark的运行原理你能不能在十五分钟内把RDD的依赖关系、宽窄依赖、stage划分讲清楚很大程度上决定了面试官对你的基础是否扎实的判断。从应用场景来说词频统计本身虽然简单但它延伸出去的业务价值很广日志分析中统计高频错误码、舆情监控中分析热词分布、电商场景中挖掘用户搜索意图、搜索引擎中构建倒排索引的统计基础。把底层原理吃透了这些场景对你来说就是换一层皮而已。提示如果你已经有单机处理万级文本的经验请先放下“这不就是一个Counter嘛”的惯性思维。分布式环境下的词频统计难点从来不在逻辑本身而在于你能否理解数据是怎么被切分、被打乱、再被重新汇总的。理解了这个过程你才算真正迈过了分布式计算的门槛。2. 设计思路与核心技术选型2.1 为什么选择Spark而不是其他方案拿到这个项目时最先要回答的问题其实是技术选型背后的理由是什么。如果你只是想在本地统计一个几MB的文件我完全推荐你用Python的collections.Counter三行代码搞定没必要引入任何分布式框架。但词频统计作为Spark的项目它的意义在于让学习者体会一套完全不同的数据思维方式。我们做个对比单机方案最大的痛点是什么内存限制。假设你有一份10GB的日志文件需要统计关键词频次单机方案通常有两种做法一次性读入内存然后OOM内存溢出教你做人或者手动分批读取、外部排序、合并中间结果——这一套流程写下来没有几百行代码根本打不住而且每一处并行化都要你自己处理线程安全和结果合并。MapReduce倒是能解决分布式问题但它的致命伤在于中间结果必须落地到磁盘每轮Map和Reduce之间的落盘开销在迭代计算场景下极其浪费。Spark聪明在什么地方呢它设计了一个弹性分布式数据集RDD让数据可以驻留在内存中中间结果尽量不落盘并且把Map和Reduce之间的shuffle过程做成了可配置、可调优的。对词频统计这种经典的Map-Reduce模式用Spark来写代码量极精简执行效率还高。简单算笔账同样的词频统计任务处理100GB文本单机方案大概率写个脚本要跑到几小时而且随时可能宕机MapReduce方案能跑通但中间落盘会拖慢大概3-5倍的速度Spark方案通过合理配置executor内存和并行度通常能控制在MapReduce耗时的四分之一左右。当然这个前提是数据能均匀分布后面讲数据倾斜的时候我们再说例外情况。2.2 RDD与DataFrame API的取舍博弈接下来说一个很多新手会纠结的问题写词频统计到底用RDD还是DataFrame我的建议是这个项目里两个都写但理解重点放在RDD上。为什么因为DataFrame的底层逻辑虽然更优化但它把很多分布式细节藏掉了。比如你写一个DataFrame的groupBy聚合Spark优化引擎会自动帮你做谓词下推、列剪枝、甚至选择更优的聚合策略这对生产开发来说是好事但对学习分布式原理来说是障碍。RDD的API则更贴近MapReduce的原生模型你能清楚地看到每个转换操作对应什么样的任务调度和数据流动。举一个实际的例子用RDD写词频统计核心链路是这样的textFile读入文件生成RDDflatMap做行转词mapToPair生成键值对reduceByKey做分组聚合。每一步你都能感知到数据在分区间的流转。而用DataFrame你只要读入、explode、groupBy三个动作就完了优化器帮你在背后做了大量处理。新手面对后者会产生“原来Spark这么简单”的错觉然后到真正调优时就懵了。当然我也要提醒你没有必要把RDD和DataFrame对立起来。真实项目中我经常用DataFrame做数据清洗和过滤到了复杂聚合逻辑再切回RDD操作。理解了各自的适用边界才是成熟工程师的做法。2.3 集群部署模式与运行环境考量另一个决定项目走向的选型是运行环境。词频统计可以在本地模式跑通也可以在Standalone集群、YARN集群、甚至Kubernetes环境上跑。本地模式适合什么情况就是你只是想验证代码逻辑是否正确数据量不大的时候。它不需要任何集群环境一个进程就搞定了在开发调试阶段效率极高。我在本地开发的时候长时间都是Master(local[*])一把梭跑通逻辑后再提交到集群去执行。到了真正体现分布式价值的阶段我建议至少在Standalone或YARN模式上跑一遍。Standalone集群模式配置相对简单适合做学习实验生产环境常用的是YARN。也都各自有支持的理由——Standalone对Spark本身的支持最直接部署简单、日志清晰YARN的优势在于多框架资源统一调度如果你所在的团队有Hadoop生态其他组件用YARN能实现集群资源共享。另外我特别想提一个很容易踩的坑本机内存非常充足代码在本地模式怎么跑都秒出结果但一上了集群executor数量、每个executor的内存大小、并行度这些参数全都变了性能表现可能急转直下。这就引出了一个问题你在本地看到的“正确性”不一定等于在集群上的“稳定性”。后面我会单独拿出来详细说。3. 环境准备与代码骨架构建3.1 版本选择与依赖配置开始写代码之前先把版本兼容问题理清楚。这是我在无数次帮人排查环境问题后总结出来的教训Spark版本、Scala版本、JDK版本之间是有严格对应关系的不是随便组合就能跑起来的。就目前的实践来看一个比较稳妥的组合是JDK 8或JDK 11配合Spark 3.x。Spark 3.2以后官方推荐使用JDK 8/11到了Spark 3.4虽然也支持JDK 17但我个人在生产环境保守起见还是用JDK 8或11居多。Scala方面Spark 3.x自身基于Scala 2.12编译你用Scala写代码时保持2.12版本就不会有兼容性麻烦。如果你选择用Python写PySpark就无所谓Scala版本问题了只要保证Python版本在Spark支持范围内即可。依赖管理工具通常用Maven或sbt。Maven上手成本更低国内的镜像源也相对稳定所以我推荐初学者用Maven来管理项目。dependencies dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.3.2/version /dependency /dependencies注意这个artifactId里的_2.12指的是Scala的版本号。提示如果你是PySpark路线直接用pip安装pyspark即可它会自动拉取对应版本的Spark二进制文件对小规模学习场景来说是最省事的方式。但生产环境一般还是用独立的Spark集群PySpark作为客户端提交任务。3.2 本地模式下的快速启动我把学习阶段的环境准备简化成三步照着走基本半小时内就能进入写代码的状态第一步下载Spark发行包。去官网下载预编译的二进制包选择带有Hadoop依赖的版本比如spark-3.3.2-bin-hadoop2.7或对应你Hadoop环境要求的版本。如果你只是本地跑不需要额外部署HadoopSpark发行包里的单机模式足够用了。第二步配置环境变量。在系统的PATH中加入Spark的bin目录方便后续直接使用spark-submit和spark-shell命令。第三步验证环境。在命令行输入spark-shell如果能看到Spark的Logo和一些启动日志说明环境已经就绪。我习惯在这个阶段跑一个最简单的Scala一行代码来确认基础功能正常spark.range(1000).count()能返回1000就说明完全没问题。等环境就绪后你还需要准备测试数据。强烈建议不要直接用英文小说做测试而是自己构造一份混合了大小写、标点符号、空行、特殊字符的脏数据。你会发现真实的文本数据比教科书里的干净例子值得处理得多。3.3 Scala与Python双语言的选择建议我在团队里带人做这个项目时经常被问到要选Scala还是Python。这里我给一个不偏不倚的建议如果你没有强烈的语言偏好我的推荐排序是——初学者选PySpark长期干大数据开发选Scala。PySpark的优势是上手极快。Python的语法对大部分人来说更亲切写起来不烧脑数据处理的思路能更直接地映射到代码上。如果你本身就做数据分析、算法相关的工作PySpark几乎是唯一合理的选择。Scala的优势则体现在生产性能和类型安全上。Spark的内核就是用Scala写的使用Spark的原生语言意味着你在遇到一些高级API或者新功能特性时能第一时间使用而不用等PySpark的适配Scala的函数式编程风格和Spark的算子设计理念天然契合很多链式操作方法在手写时非常顺滑Scala是静态类型语言很多低级的类型错误在编译期就能暴露出来这对于一个几十上百行的数据处理逻辑来说能省掉不少调试时间。我的建议是这个项目你可以用PySpark先跑通一遍理解整体逻辑如果精力允许再用Scala原样实现一遍对比两次写代码的体验差异。这个过程本身对编程能力的提升很有帮助——语言转换翻译的过程中你对逻辑本身的理解会变得异常清晰。4. 词频统计核心代码实现与执行原理4.1 基于RDD的完整实现方案我先把Scala版本的RDD核心代码完整贴出来然后我们逐段拆解。这段代码我做了足够的简化保留了最核心的逻辑链路。import org.apache.spark.{SparkConf, SparkContext} object WordCount { def main(args: Array[String]): Unit { val conf new SparkConf() .setAppName(WordCountDemo) .setMaster(local[*]) val sc new SparkContext(conf) val inputPath data/input/words.txt // 读取文本文件并逐行处理 val lines sc.textFile(inputPath) // 每行拆分为单词 val words lines.flatMap(_.split( )) // 清洗转小写 去掉标点符号简化版 val cleanedWords words.map(_.toLowerCase.replaceAll([^a-zA-Z\\s], )) // 构造 (单词, 1) 键值对 val wordPairs cleanedWords.map(word (word, 1)) // 按单词聚合求和 val wordCounts wordPairs.reduceByKey((a, b) a b) // 按出现次数降序排列 val sortedCounts wordCounts.sortBy(_._2, ascending false) // 输出结果 sortedCounts.collect().foreach(println) sc.stop() } }代码核心逻辑确实不长但要真正理解它背后的执行过程你得看到数据在这些算子之间是怎么流动的。我用分步的方式拆给你看。先从sc.textFile说起。这个操作会创建一个HadoopRDD底层根据文件的大小和配置的分区数把文件拆成多个分区。分区数默认情况下等于文件块的大小映射关系你在本地模式下可能只有两三个分区但在集群上一个128MB的块可能就是一个分区。这一步是整个处理链条的入口决定了后续所有算子的并行度基数。接着是flatMap(_.split( ))。flatMap的语义是“先map后展平”对每个输入元素做映射后把映射结果为集合的结果展开成一个个独立的元素。举个例子一行文本“hello world hello”经过split后得到数组[hello, world, hello]flatMap会把它变成三个独立的单词hello、world、hello。注意这里的“独立”是一个逻辑概念实际数据可能分布在多个分区每个分区各自计算。然后是map操作把每个单词转换成单词1的形式。到这里为止所有的算子都是窄依赖——每个分区的输入数据只流向固定的输出分区没有跨节点的数据交换。这个阶段在Spark里被称为“流水线执行”多个窄依赖的转换会合并到一个stage中一气呵成。关键点在reduceByKey。这个算子在执行时涉及一次shuffle。具体过程是每个分区中hello, 1和hello, 1会先在本地做一次合并变成hello, 2这就是map-side预聚合也叫combiner。然后相同key的数据根据哈希规则被分到同一个下游分区进入reduce阶段做最终求和。设计一个简单的计算公式帮你理解分区策略目标分区编号 hash(key) % 下游分区数。默认的hash算法比较简单但其核心思路是确保“相同key进同一个分区”。最后是sortBy它会触发一次全量排序这一步也会产生shuffle。如果你不需要排序完全可以省略这一步直接输出能省下不少时间。4.2 基于DataFrame的对比实现RDD版本理解到位以后再看DataFrame版本就显得很清爽了。我用PySpark代码演示方便对比两种语言风格的差异。from pyspark.sql import SparkSession from pyspark.sql.functions import explode, split, lower, regexp_replace, col spark SparkSession.builder \ .appName(WordCountDataFrame) \ .master(local[*]) \ .getOrCreate() df spark.read.text(data/input/words.txt) words_df df.select( explode(split(lower(col(value)), \\s)).alias(word) ) cleaned_df words_df.select( regexp_replace(col(word), [^a-zA-Z], ).alias(word) ).filter(col(word) ! ) word_counts cleaned_df.groupBy(word).count().orderBy(count, ascendingFalse) word_counts.show(50) spark.stop()代码直观了很多。你其实看不到flatMap、reduceByKey这些算子但它们以另外的名字存在于优化计划中。Spark SQL引擎会把这个逻辑翻译成物理执行计划底层操作和RDD版本的逻辑很相似但优化器会根据数据的统计信息自动选择更高效的执行策略。这里我想认真提醒一点DataFrame的groupBy在遇到大量不同key的小数据量场景时优化效果很好但在少数几个高频词占据统治地位的场景下没错说的就是某些日志中某个错误码出现率高达90%以上同样可能出现数据倾斜问题。优化器并不能帮你解决数据分布本身的不均衡它只能保证在标准场景下帮你选一条相对高效的执行路径。所以我的经验是理解RDD版本能让你看清困局产生的原因使用DataFrame版本能让你在大多数常规场景下更高效地完成开发。两者不是对立的而是不同抽象层级之间的互补。4.3 每个阶段的执行机制深度解读很多人对“stage”和“job”的划分概念就是看文档一扫而过真正遇到性能问题或者要排查任务卡住时对这些概念的理解深度就完全暴露了。我结合词频统计这个项目把执行机制的关键节点串一遍。当你调用collect()或者saveAsTextFile()这类触发行动的操作时Spark会构建一个DAG有向无环图来描述计算步骤。对上面的RDD代码来说这个DAG大致长得是这个样子的textFile - flatMap - map - reduceByKey - sortBy - collect。DAG上每遇到一个宽依赖典型如reduceByKey、sortBy中的shuffle就会在此处划分出一个新的stage。上面的链路实际上会被拆成三个阶段第一个阶段从textFile到map结束的窄依赖流水线第二阶段是reduceByKey的shuffle过程和后续的聚合计算第三阶段是sortBy的shuffle演变出来的排序计算。每个stage内部包含多个可以并行执行的tasktask的数量由分区的数量决定。用形象的类比来说可以把RDD理解成一条流水线上传输的工件窄依赖就是在同一条流水线上给工件做打磨、贴标签而宽依赖就像工件需要被分拣到不同车间的过程——车间之间还得先把工件打包运过去这个运输过程就是shuffle。shuffle这个环节在分布式计算里几乎所有的性能问题都与之有关。一次shuffle涉及每个父分区数据的序列化、落盘、网络传输、反序列化、再聚合。词频统计这种key数量极大的场景shuffle数据量可能达到原始数据的数倍而且如果某个单词出现频率极高它的中间数据就可能集中涌向一个下游分区导致“一个人干活其他人围观”的数据倾斜困局。我在实际项目中看过太多教训了外部看起来任务跑不动点开Spark UI一看某个stage有几十个task绝大多数几百毫秒跑完一两个task卡了半小时都不动——九成是在shuffle上出了问题。词频统计虽然简单但它给你制造了一个极好的观察shuffle各环节的窗口你完全可以在Spark UI的stage详情页看到shuffle read和shuffle write的真实数据量。这个项目最宝贵的价值其实就在这些地方。5. 从单机脚本走向集群部署5.1 本地模式与集群模式的行为差异我见过不少开发者在本地能跑通的代码提交到集群上却问题百出。这不是因为他们水平差而是因为本地模式的执行环境和集群模式之间的差异太大了。本地模式有一个容易让人产生错觉的特征它在单个JVM进程中模拟多线程并行虽然可以配置多个executor核但本质上所有数据都走的是本地内存和本地磁盘没有真正的网络传输。因此你对shuffle的感知几乎为零——一切都在毫秒级完成。到了真正的集群环境网络带宽、磁盘IO、节点的内存限制都会成为瓶颈一次shuffle可能要搬运几个GB的数据执行时间从秒级跳到分钟级甚至小时级。另一个容易踩坑的差异是文件系统。本地模式下textFile读取的是本地路径提交到集群后你通常需要把数据放到HDFS或者对象存储上。如果不注意修改路径任务会直接报文件找不到有些人会在这个问题上反复纠结迟迟以为是代码逻辑错误。再者本地模式下setMaster(local[*])就够了但提交到集群环境这个配置通常会被提交命令中的master参数覆盖。如果不理解这个覆盖关系你可能在集群上跑了半天实际上还是在每个节点的本地模式下运行你的JVM里。5.2 打包与提交命令的完整解析集群部署的第一步是打包。Maven项目直接执行mvn clean package -DskipTests就能生成jar包。如果你的项目使用Spark的Scala相关依赖打包时注意不要把所有依赖都打进去——Spark环境本身已经有这些库了重复打包反而可能引起版本冲突。打好的jar包需要通过spark-submit提交。这是我经常用的一条命令模板给你做个参考spark-submit \ --class com.example.WordCount \ --master yarn \ --deploy-mode cluster \ --driver-memory 2g \ --executor-memory 4g \ --executor-cores 2 \ --num-executors 4 \ spark-wordcount-1.0-SNAPSHOT.jar \ hdfs:///input/words.txt我来逐一解释每个参数的作用。--class指定主类名就是你写了main方法的那个对象。--master指定资源管理器地址生产环境常用yarn。--deploy-mode cluster表示把driver也放到集群上运行日志不会直接打到你的终端如果选client模式driver跑在提交端方便看到日志但会占用提交机的资源。--driver-memory和--executor-memory分别控制驱动器和执行器内存这个值要参考每个执行器上实际需要处理的数据量来定给少了会频繁GC或OOM给多了会浪费集群资源甚至申请不到资源。--executor-cores和--num-executors合起来决定了spark的并行计算能力简单的经验公式executor总核数等于两者乘积建议不超过集群总核数的70%留出系统余量。最后的参数是程序入口接收的命令行参数在代码里通过args[]获取并作为输入路径。我前面给出的Scala代码里输入路径是硬编码的真实项目肯定要改成参数传入的方式。5.3 资源参数估算的实用经验资源参数不是拍脑袋定的我给你一个可操作的估算思路。假设输入数据是100GB的文本歪曲一点看经过flatMap和map之后RDD中每个元素从一行变为了单词1的键值对数据总量可能会有30%-50%的膨胀。如果每个executor分到4GB内存那么执行过程中的数据驻留和shuffle缓冲区都要从这里面扣出空间一般建议保留30%余量给Spark内部管理层。以一个executor 4核、4GB内存为例保守估算它能在内存中稳定处理的数据量大概在1-2GB之间。你需要把它视为一个可调基准线然后通过试跑不断修正。还有一个常被忽视的参数是spark.default.parallelism它决定了shuffle之后默认的RDD分区数。不设置时Spark会根据上游分区数推断但这个推断不一定为你当前的场景最优。词频统计这种场景中我建议把这个参数设置为executor总核数的2-3倍能比较有效地压满CPU又不会造成过多的小任务调度开销。提示无论你怎么估算第一轮上线都建议先用小数据集跑通大致估算单分区处理速率再按比例推算全量数据的参数。盲目照模板设参和过度调参都不好。6. 常见问题与性能调优实录6.1 数据倾斜与Skew问题处理处理大数据集时遇到数据倾斜的概率非常高这个词频统计项目也不例外。我举个例子你统计一部小说的词频其中冠词“the”出现的频率可能是其他单词的几十上百倍。当reduceByKey执行时所有包含“the”的键值对会进入同一个分区负责那个分区的task要处理的数据量远超其他分区于是它变成了整个任务的短板其他task早跑完了只能干等着它。怎么定位数据倾斜经验法则是看Spark UI中某个stage的各task耗时分布如果看到明显的长尾效应——大量task耗时在秒级个别task耗时在分钟级且shuffle read数据量也呈现同样分布基本可以确定是倾斜问题。常见的解决策略我整理了一个表方案适用场景操作思路提高shuffle分区数倾斜不严重通过spark.sql.shuffle.partitions或repartition增加分区让单个分区数据量更小两阶段聚合倾斜集中在少数key先加上随机前缀分散到更多分区做局部聚合去前缀后再做全局聚合过滤异常key倾斜由少数无效脏数据导致对无意义的高频词如空串、单字符先行过滤广播map小表与倾斜大表join场景不适用于纯词频统计但方法通用性很广两阶段聚合是词频统计里能直接落地的方案我展开说一下思路。第一轮reduceByKey之前给每个key加一个随机前缀比如0到N之间的随机数这样原本相同的key会被打散到N个不同的分区每个分区都做一次局部聚合。完成之后去掉前缀再做一次reduceByKey合并。代价是执行两轮shuffle但换来了负载均衡当少数key倾斜严重时收益远超开销。6.2 初学阶段的高频报错与排查思路我整理了一些初学者在跑词频统计时最常撞见的错误每个都是我实际带人时遇到过的第一个错误SparkContext或SparkSession already stopped。原因是你可能手动调用了stop()但代码后面又尝试继续用这个上下文执行操作。解决办法很简单确保stop()放在所有计算完成之后而且不要在循环中重复创建多个上下文。第二个错误Exception in thread main java.lang.OutOfMemoryError: Java heap space。常见诱因是读取的文件过大分区数太少导致单个分区数据量超出executor内存。解决方向有两种调大executor内存或者增加分区数来降低单分区数据量。如果是本地模式还可以调大driver内存来缓解。第三个错误找不到主类或打包体结构不对。无非就是class名写错、jar内部MANIFEST不完整、或者用错Spark版本导致Scala兼容性报错。逐一排查不要急躁。第四个错误路径错误。本地模式路径写file:///前缀集群环境写hdfs://或s3a://前缀。经常有人在集群上跑还指着一个本地路径跑偏任务一直报输入文件不存在。还有个比较隐蔽的坑flatMap在做单词拆分时直接split( )对连续多个空格处理出的空字符串没有做过滤导致计数结果里出现一堆空串或者纯标点符号项。这种数据质量问题在真实项目里影响不大不小容易被忽视但频率很高。我通常会在拆分成单词后加一个filter(_.nonEmpty)来做兜底。6.3 提速技巧从代码细节到参数调优最后分享几个真刀真枪跑调优时的细节经验。第一个技巧减少shuffle的数据量。词频统计里reduceByKey本身自带map-side聚合combiner你要确认这个特性没有被关闭或误改。如果代码框架里用了groupByKey来替代reduceByKey——恰恰是很多从MapReduce转过来的人容易犯的错误——那么每个key对应的所有原始值都会被原封不动地传到reduce端数据量大了数倍。绝对不要用groupByKey做聚合永远优先考虑reduceByKey或aggregateByKey。第二个技巧序列化方式优选Kryo。Spark默认的Java序列化器性能较差可以通过配置切换为Kryo对纯字符串的键值对数据能获得更小的序列化体积和更快的速度conf.set(spark.serializer, org.apache.spark.serializer.KryoSerializer)第三个技巧如果与上游对接的是Parquet格式的数据尽量基于DataFrame API处理并开启向量化读取如果数据是纯文本则考虑是否可以先做一次预处理转换成Parquet再进入后续统计流程。第四个技巧留意Spark 3.0以后引入的动态资源分配功能。在集群环境配置得当的前提下它能根据任务负载自动调整executor数量显著降低空闲资源浪费。但要注意它适合长跑型任务对短任务反而会产生额外的调度开销。第五个技巧是关于缓存策略的。词频统计这类单轮任务其实不太需要任何缓存。但如果你要把中间结果复用在多个后续计算中就要考虑persist的存储级别——数据量不大用MEMORY_ONLY数据量大会出现频繁GC就用MEMORY_AND_DISK_SER。cache()只是persist的一个便捷方法的别名用的存储级别是MEMORY_ONLY对于较大的RDD这是风险很高的选择。我把这些技巧的优先级排个序放在这供你按需选择最高优先级用reduceByKey替代groupByKey高优先级合理设置并行度和executor内存中优先级开启Kryo序列化视情况两阶段聚合处理倾斜场景、缓存与复用策略词频统计项目规模不大但它能让你看清分布式计算原理的全貌。我个人在实际操作中体会最深的点是很多问题只有亲手把一个小项目放到Spark UI面前仔细观察才会真正明白调优参数的意义。这个项目做完以后你可以顺带尝试一些扩展方向把中文分词加入统计流程、把结果输出到Parquet格式供下游分析、或者做一份带时间窗口的热词趋势分析。每个扩展都会逼着你碰到新的分布式问题那才是成长最快的时候。