1. 从一个看起来简单的WordCount说起但凡接触过分布式计算的人几乎都绕不开WordCount这个案例。它就像编程语言里的Hello World看着简单但真正动手写的时候你会发现里面藏着不少门道。我见过太多人第一次跑WordCount代码照着示例敲完提交上去结果要么报错要么输出一堆空文件要么结果跟预期对不上。问题往往不在逻辑本身而在于对Spark运行机制的理解不够深。这篇内容我打算把WordCount和TopKey这两个经典案例彻底讲透。不只是给你一份能跑的Scala代码而是把每个算子为什么这么选、参数为什么这么设、集群上跑的时候会遇到什么坑全部掰开揉碎说清楚。适合已经了解Scala基础语法、想真正把Spark用起来的人。如果你还在纠结要不要学Spark那这篇可能不太适合你但如果你已经决定要上手并且想少走弯路那接下来的内容应该能帮你省下不少调试时间。先明确一下这两个案例的核心价值。WordCount解决的是词频统计问题本质上是把一堆文本拆成单词然后按单词分组计数。TopKey则是在WordCount的基础上找出出现频率最高的前N个词。这两个案例覆盖了Spark核心编程中最常用的算子组合flatMap、map、reduceByKey、sortBy、take等。把它们吃透后面写更复杂的业务逻辑就是搭积木的事。2. 环境准备与项目骨架搭建2.1 依赖版本的选择逻辑在动手写代码之前先把环境理清楚。Spark的版本迭代很快不同版本之间API有差异尤其是Spark 2.x和3.x之间。我个人的建议是如果是新项目直接上Spark 3.x因为它在性能优化和API简洁性上都有明显提升。但如果你所在的环境还在用Spark 2.x那也没关系WordCount和TopKey这两个案例的核心API在两个版本里基本一致。Scala版本方面Spark 3.x默认用的是Scala 2.12Spark 2.4之前用的是2.11。这个必须对齐否则编译的时候会报NoSuchMethodError之类的错误而且这种错误往往在运行时才暴露排查起来很烦。我踩过这个坑当时本地Scala是2.13集群上是2.12代码本地跑得好好的一提交就挂查了半天才发现是版本不匹配。构建工具用Maven或者sbt都行。sbt对Scala项目更友好依赖管理更自然Maven的话配置稍微繁琐一点但团队协作时更通用。下面以Maven为例给出核心依赖配置dependencies dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.3.0/version /dependency dependency groupIdorg.scala-lang/groupId artifactIdscala-library/artifactId version2.12.15/version /dependency /dependencies注意artifactId里的_2.12后缀这个数字必须和你的Scala版本一致。很多人复制依赖的时候只看版本号忽略了后缀结果编译报错找不到类。2.2 项目目录结构目录结构不用太复杂但建议按职责分一下。我习惯这样组织src/main/scala/ com/example/spark/ WordCount.scala TopKey.scala util/ SparkSessionBuilder.scala把创建SparkSession的逻辑单独抽出来是因为实际项目中几乎每个任务都需要初始化SparkSession而且配置参数往往有共性。抽成一个工具类后面写新任务的时候直接调用省得重复写一堆配置。package com.example.spark.util import org.apache.spark.sql.SparkSession object SparkSessionBuilder { def create(appName: String, master: String local[*]): SparkSession { SparkSession.builder() .appName(appName) .master(master) .config(spark.sql.shuffle.partitions, 10) .getOrCreate() } }这里我特意设了spark.sql.shuffle.partitions为10。默认值是200对于小数据集来说200个分区意味着200个任务每个任务处理的数据量极小调度开销反而成了瓶颈。本地测试的时候设小一点能明显感觉到速度提升。当然生产环境要根据数据量调整这个后面会细说。3. WordCount的Scala实现与算子拆解3.1 完整代码与逐行解读先上完整代码然后逐段拆解package com.example.spark import com.example.spark.util.SparkSessionBuilder object WordCount { def main(args: Array[String]): Unit { val spark SparkSessionBuilder.create(WordCount) val sc spark.sparkContext val inputPath args(0) val outputPath args(1) val textRDD sc.textFile(inputPath) val wordCounts textRDD .flatMap(line line.split(\\s)) .filter(_.nonEmpty) .map(word (word, 1)) .reduceByKey(_ _) wordCounts.saveAsTextFile(outputPath) spark.stop() } }这段代码看着简单但每一行都有讲究。sc.textFile(inputPath)读取文件返回的是一个RDD[String]每个元素是文件中的一行。这里有个细节如果输入路径是本地文件系统需要确保集群上每个节点都能访问到如果是HDFS路径那就没这个问题。我见过有人在本地用file:///路径测试提交到集群后报FileNotFoundException就是因为路径没改成HDFS。flatMap(line line.split(\\s))把每行拆成单词。用\\s而不是简单的空格是因为实际文本里可能有多个连续空格、制表符、换行符等。flatMap和map的区别在于flatMap会把拆分后的数组拍平返回一个包含所有单词的RDD而不是RDD[Array[String]]。filter(_.nonEmpty)过滤掉空字符串。这个步骤容易被忽略但如果文本里有连续空格split之后会产生空字符串不过滤的话最后统计结果里会出现一个空key看着很别扭。map(word (word, 1))把每个单词映射成键值对这是为后面的reduceByKey做准备。Spark的reduceByKey要求输入是(key, value)形式的RDD。reduceByKey(_ _)按key分组对value做累加。这里用reduceByKey而不是groupByKey是因为reduceByKey会在map端做本地聚合减少shuffle的数据量。对于词频统计这种场景reduceByKey的性能优势非常明显。3.2 reduceByKey和groupByKey的性能差异很多人写WordCount的时候习惯用groupByKey然后mapValues(_.sum)逻辑上没问题但性能差很多。我做过一个测试同样的数据集reduceByKey比groupByKey快了将近三倍。原因在于shuffle机制。groupByKey会把所有相同key的value拉到同一个分区不做任何预聚合。如果某个key对应的数据量特别大还会导致数据倾斜某个task处理的数据量远超其他task拖慢整个作业。reduceByKey则会在map端先做一次本地聚合相当于把局部求和提前做了shuffle的时候只传输聚合后的结果数据量小很多。注意如果业务逻辑不是简单的求和而是需要保留所有原始value比如要计算中位数那groupByKey还是得用。但纯粹做计数、求和这类聚合操作优先选reduceByKey。3.3 输出结果的处理与常见问题saveAsTextFile(outputPath)把结果写到指定路径。这里有几个坑第一输出路径不能已经存在。Spark会检查路径如果目录已存在直接抛异常。调试的时候反复运行每次都要手动删输出目录很烦。可以在代码里加一段判断或者用hadoop fs -rm -r先清理。第二输出是一个目录里面包含多个part文件数量等于最终RDD的分区数。如果分区数是200就会生成200个文件哪怕数据量很小。这时候可以用coalesce(1)把结果合并到一个文件里但要注意coalesce(1)会把所有数据拉到一个节点数据量大时可能OOM。更稳妥的做法是repartition(1)但同样有性能问题。我的建议是小结果集用coalesce(1)大结果集保持多文件输出后续用其他工具合并。第三输出格式是(word, count)的字符串形式因为saveAsTextFile调用的是元素的toString方法。如果后续要程序化读取这种格式不方便解析。可以考虑用saveAsObjectFile或者转成DataFrame后写CSV/Parquet。4. TopKey的实现思路与排序策略4.1 从WordCount到TopKey的演进TopKey的需求是在词频统计的基础上找出出现次数最多的前N个词。这个需求在实际场景中很常见比如分析热搜词、统计高频错误码、找出访问量最大的URL等。实现思路有两种。第一种是先做完整的WordCount然后对结果按count降序排列取前N个。第二种是用takeOrdered或者top算子直接在RDD层面做。两种方式各有适用场景下面分别说。4.2 基于sortBy的全局排序方案先看第一种也是最直观的方案package com.example.spark import com.example.spark.util.SparkSessionBuilder object TopKey { def main(args: Array[String]): Unit { val spark SparkSessionBuilder.create(TopKey) val sc spark.sparkContext val inputPath args(0) val topN args(1).toInt val textRDD sc.textFile(inputPath) val topKeys textRDD .flatMap(_.split(\\s)) .filter(_.nonEmpty) .map(word (word, 1)) .reduceByKey(_ _) .sortBy(_._2, ascending false) .take(topN) topKeys.foreach(println) spark.stop() } }这段代码的核心是sortBy(_._2, ascending false)。sortBy会对整个RDD进行全局排序底层通过rangePartitioner实现。排序完成后take(topN)取前N个元素。这里有个性能问题sortBy是全局排序会触发一次全量shuffle把所有数据按照count重新分区。如果数据量很大这一步开销不小。而且我们只需要前N个却对所有数据做了排序有点浪费。4.3 takeOrdered与top的取舍Spark提供了takeOrdered和top两个算子专门用于取前N个元素不需要全局排序。它们的区别在于top(n)取最大的N个元素使用默认的降序排列。takeOrdered(n)取最小的N个元素使用默认的升序排列。对于TopKey场景我们要取count最大的N个词可以用top但top比较的是整个元素而我们的元素是(word, count)这样的元组。元组的默认排序是先比第一个元素再比第二个。我们想按count排序就需要提供一个自定义的Ordering。val topKeys textRDD .flatMap(_.split(\\s)) .filter(_.nonEmpty) .map(word (word, 1)) .reduceByKey(_ _) .top(topN)(Ordering.by(_._2))Ordering.by(_._2)指定按照元组的第二个元素也就是count来比较。top内部会先在每个分区取前N个然后汇总效率比全局排序高很多。但top有一个限制它返回的是一个数组数据量大的时候这个数组会全部拉到Driver端。如果N很大Driver端可能内存不够。所以top适合N比较小的场景比如取前10、前100。如果N是几千甚至几万还是用sortBy然后take更稳妥或者用sortBy后limit再写文件。4.4 处理数据倾斜的实战技巧TopKey场景下数据倾斜是一个绕不开的问题。如果某个词出现频率极高比如的、是这类停用词那么reduceByKey的时候这个key对应的数据量会远超其他key导致某个task成为瓶颈。我遇到过一个案例日志分析任务里某个错误码占了总日志量的40%reduce阶段直接卡住。解决办法有几个第一预处理阶段过滤掉明显的停用词。如果是文本分析维护一个停用词表在flatMap之后加一个filter把停用词去掉。这个最简单效果也最直接。第二加盐打散。给key加上随机前缀比如(word _ Random.nextInt(10), 1)先做一次聚合然后再去掉前缀做第二次聚合。这样能把一个热key拆成10个分散到不同task上。代价是多了一次shuffle但能解决倾斜问题。第三提高并行度。增加shuffle的分区数让每个task处理的数据量小一点。可以通过spark.sql.shuffle.partitions或者reduceByKey(_ _, numPartitions)来指定。提示加盐方案会增加代码复杂度如果不是倾斜特别严重优先考虑过滤停用词和调整并行度。加盐适合倾斜非常极端、且无法通过过滤解决的场景。5. 集群提交与参数调优的实操细节5.1 spark-submit的常用参数代码写好了本地跑通只是第一步真正要上集群得用spark-submit提交。下面是一个典型的提交命令spark-submit \ --class com.example.spark.WordCount \ --master yarn \ --deploy-mode cluster \ --executor-memory 4g \ --executor-cores 2 \ --num-executors 10 \ --driver-memory 2g \ wordcount.jar \ hdfs:///input/text.txt \ hdfs:///output/wordcount每个参数都有讲究。--executor-memory决定每个Executor的堆内存--executor-cores决定每个Executor能并行跑几个task。--num-executors是Executor的数量。这三个参数共同决定了作业的总并行度num-executors * executor-cores。很多人调优的时候只盯着内存加忽略了cores。实际上如果cores太少每个Executor只能串行跑task内存再大也用不上。一般来说每个Executor配2到5个cores比较合适太少浪费内存太多会导致GC压力大。--driver-memory容易被忽略。如果代码里有collect、take这类把数据拉到Driver端的操作Driver内存不够会直接OOM。TopKey场景下如果用top取前N个N很大时就要注意Driver内存。5.2 内存配置的计算逻辑Executor内存怎么定一个粗略的估算方法是总数据量除以总并行度再乘以一个安全系数。假设输入数据是10GBnum-executors10executor-cores2那么总并行度是20。每个task平均处理500MB数据。考虑到shuffle、缓存、对象开销每个task至少需要1GB内存那么每个Executor跑2个task需要2GB用于计算再加上堆外内存和系统预留executor-memory设4GB比较稳妥。当然这只是估算实际还要看数据特征。如果数据里字符串特别多对象开销会更大如果有大量shuffle内存需求也会上升。我的经验是先按估算值跑一次看Spark UI里的GC时间和spill情况再针对性调整。5.3 提交后常见报错与排查提交到集群后最常见的报错有这么几类第一类ClassNotFoundException。通常是依赖没打进去。用Maven的shade插件打一个fat jar把所有依赖都包含进去。注意Spark本身的依赖要设成provided因为集群上已经有了打进去反而可能冲突。第二类Container killed by YARN for exceeding memory limits。这是Executor内存超了。可能是数据倾斜导致某个task处理的数据量特别大也可能是内存配置本身就不够。先看Spark UI里各task的处理数据量确认是不是倾斜再决定是加内存还是解决倾斜。第三类Shuffle fetch failed。通常是Executor挂了导致shuffle数据丢失。可以开启spark.shuffle.service.enabled让shuffle数据由外部服务管理Executor挂了也不影响。另外spark.network.timeout可以适当调大避免网络抖动导致的fetch失败。第四类输出目录已存在。前面提过Spark不允许输出到已存在的目录。可以在代码里加判断或者每次提交前手动清理。6. 从案例到生产的思维转变6.1 本地模式与集群模式的差异本地模式跑通不代表集群上没问题。我总结了几点差异都是实际踩过的第一文件路径。本地用file:///集群上要用HDFS路径。如果代码里写死了本地路径提交上去必然报错。建议把路径作为参数传入不要硬编码。第二序列化。本地模式下很多对象不需要序列化就能传递集群模式下所有在Executor上执行的对象都必须可序列化。如果代码里用了自定义类记得实现Serializable接口。否则会报NotSerializableException而且报错信息往往指向一个看起来无关的地方排查起来很费劲。第三日志级别。本地跑的时候控制台输出很清晰集群上跑日志分散在各个Executor的container里需要去YARN的日志聚合里看。建议在代码里把日志级别调成WARN减少不必要的输出方便定位问题。6.2 代码健壮性的几个习惯写Spark代码有几个习惯能帮你省很多事习惯一参数校验。输入路径、输出路径、TopN这些参数在main函数开头就校验不合法直接退出别等到运行到一半才报错。习惯二资源关闭。spark.stop()要确保执行最好放在finally块里。虽然SparkSession不关也不会导致严重问题但长期运行的服务里不关会占着资源。习惯三异常捕获。对于可能失败的操作比如读写HDFS加try-catch把异常信息打出来方便排查。但不要吞异常捕获了要处理或者重新抛出。习惯四配置外置。把路径、并行度、内存这些配置抽到配置文件或者命令行参数里不要硬编码。这样换环境的时候不用改代码重新提交就行。6.3 性能监控与调优的切入点作业跑起来之后怎么知道性能好不好看Spark UI。重点看几个指标Task的duration分布如果大部分task很快少数task特别慢说明有数据倾斜。GC时间如果GC时间占比超过10%说明内存不够需要加内存或者减少缓存。Shuffle read/write如果shuffle数据量特别大考虑优化算子减少shuffle。Spill如果出现spill说明内存不足以容纳当前分区的数据需要加内存或者增加分区数。调优是一个迭代的过程没有一劳永逸的配置。我的做法是先保证功能正确再根据监控数据逐步优化。每次只调一个参数观察效果避免同时改多个参数导致无法定位问题。7. 一些容易被忽略的细节7.1 文本编码问题读取文本文件时默认是UTF-8编码。如果文件是GBK编码中文会出现乱码。解决办法是在读取时指定编码或者先用工具转成UTF-8。sc.textFile本身不提供编码参数可以用sparkContext.hadoopConfiguration设置或者用wholeTextFiles读取后手动解码。7.2 空文件与空分区的处理如果输入目录里有空文件textFile会正常读取但对应的分区里没有数据。这本身不是问题但如果后续有mapPartitions之类的操作要处理空迭代器的情况否则可能报错。7.3 结果验证的方法跑完WordCount怎么验证结果对不对我的做法是先用一个小样本数据手动算一遍预期结果然后跟Spark的输出对比。确认逻辑没问题后再上全量数据。全量数据跑完后可以抽样检查几个key的count跟原始数据里的出现次数对一下。对于TopKey验证方法是先跑完整的WordCount把结果按count排序取前N个跟TopKey的输出对比。如果一致说明TopKey的逻辑没问题。7.4 关于Scala的隐式转换Scala的隐式转换在Spark代码里用得很多比如reduceByKey需要(K, V)类型的RDD而map返回的是普通RDD中间就涉及隐式转换。如果import不对编译会报错。常见的import包括org.apache.spark.SparkContext._它包含了这些隐式转换。Spark 2.x之后很多隐式转换移到了org.apache.spark.rdd.RDD的伴生对象里import方式有变化需要注意版本差异。8. 从这两个案例延伸出去WordCount和TopKey虽然简单但它们覆盖的算子组合可以延伸到很多场景。比如把单词换成URL就是访问量统计把单词换成错误码就是错误分布分析把count换成金额就是销售额聚合。核心模式是一样的拆分、映射、聚合、排序。掌握了这个模式再去看Spark的其他算子比如aggregateByKey、combineByKey、foldByKey会发现它们只是聚合方式不同底层逻辑是相通的。reduceByKey要求聚合前后的类型一致aggregateByKey允许聚合前后类型不同combineByKey则更灵活可以指定初始创建、分区内合并、分区间合并三个函数。我在实际项目中用得最多的是reduceByKey和aggregateByKey。前者适合简单的求和、计数后者适合求平均值、求TopN这类需要中间状态的聚合。比如求每个用户的平均消费金额就可以用aggregateByKey初始值是(0.0, 0)分区内合并是累加金额和次数分区间合并也是累加最后再相除。最后分享一个小技巧写Spark代码的时候尽量用reduceByKey替代groupByKey用mapPartitions替代map当需要在每个分区内做初始化操作时用filter尽早过滤掉不需要的数据。这些习惯能显著提升作业性能尤其是在数据量大的时候。