1. 从一个WordCount说起为什么它成了大数据入门的“必修课”但凡接触过分布式计算的人几乎都绕不开WordCount这个例子。它就像编程语言里的“Hello World”看似简单却把分布式计算最核心的几个概念——数据分片、并行处理、局部聚合、全局汇总——全都串了一遍。我当年第一次跑通WordCount的时候盯着控制台输出的词频统计愣了半天不是因为结果多复杂而是突然意识到原来一批数据可以被切成很多块分给不同的计算单元同时处理最后再合并起来这个过程对使用者几乎是透明的。这次要聊的就是围绕WordCount和它的进阶变体TopKey用Scala在Spark上完整实现一遍。WordCount解决的是“每个词出现了多少次”而TopKey解决的是“出现次数最多的前N个词是哪些”。前者是基础后者是实际业务里更常见的需求——比如统计热门搜索词、分析日志里高频访问路径、找出销量最高的商品品类等等。两个案例放在一起基本能把Spark核心RDD操作里最常用的那批算子覆盖个七七八八。这篇文章适合谁看如果你刚学Spark看了一堆概念但不知道从哪下手写代码那这两个案例就是最好的切入点。如果你已经会写WordCount但每次写TopKey都要翻文档查排序怎么写那这篇也能帮你把思路理顺。我会把代码拆开揉碎讲清楚每一步为什么这么写、有没有更优的写法、实际跑的时候容易在哪儿卡住。Scala版本用的是2.12Spark用的是3.x这些版本差异在代码里影响不大但我会在涉及的地方点出来。2. 动手之前环境准备与项目骨架搭建2.1 依赖配置与版本选择先把环境搭起来。我用的是Maven来管理依赖如果你习惯sbt也完全没问题核心依赖就那几个。Scala版本选2.12.xSpark选3.3.x或3.4.x都行这两个版本对Scala 2.12的支持都很稳定。dependencies dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.4.1/version /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-sql_2.12/artifactId version3.4.1/version /dependency /dependencies这里有个细节值得说一下spark-core和spark-sql的版本号必须完全一致Scala的二进制版本后缀_2.12也必须和你的Scala编译版本匹配。我见过不少人因为版本对不上跑起来报NoSuchMethodError排查半天以为是代码问题其实就是依赖没对齐。另外如果你是在本地IDE里跑建议把spark-core的scope设成provided打包上集群的时候不会把Spark本身的jar打进去避免和集群环境冲突。2.2 项目目录结构与代码入口目录结构没什么特别的标准的Maven布局就行src/main/scala/com/example/spark/ WordCount.scala TopKey.scala src/main/resources/ words.txt入口方面我习惯每个案例单独写一个Object带main方法方便本地直接运行调试。SparkSession的创建方式在Spark 2.0之后统一了不再需要单独创建SparkContext直接用SparkSession.builder()就行。有一点要注意本地运行时setMaster(“local[*])提交到集群时不要写死在代码里用spark-submit的–master参数指定。我一般会在代码里判断一下如果master没设置才用local这样本地和集群都能兼顾。val spark SparkSession.builder() .appName(WordCountDemo) .master(local[*]) .getOrCreate() val sc spark.sparkContext sc.setLogLevel(WARN)setLogLevel这行建议加上不然控制台会被INFO级别的日志刷屏调试的时候找自己的输出很费劲。3. WordCount完整实现从读取到输出的每一步3.1 核心代码逐行拆解先上完整代码然后我逐段解释package com.example.spark import org.apache.spark.sql.SparkSession object WordCount { def main(args: Array[String]): Unit { val spark SparkSession.builder() .appName(WordCountDemo) .master(local[*]) .getOrCreate() val sc spark.sparkContext sc.setLogLevel(WARN) val inputPath src/main/resources/words.txt val outputPath output/wordcount val result sc.textFile(inputPath) .flatMap(_.split(\\s)) .map(word (word, 1)) .reduceByKey(_ _) .sortBy(_._2, ascending false) result.collect().foreach(println) result.saveAsTextFile(outputPath) spark.stop() } }这段代码一共就十来行但每一行都值得说清楚。textFile读取文件返回的是一个RDD[String]每个元素是文件里的一行。这里有个隐含的知识点Spark读取本地文件时如果路径是本地路径它会从每个Executor的本地文件系统读这在集群模式下会出问题——因为文件可能只存在于Driver节点上。所以生产环境一般用HDFS路径或者S3路径。本地测试用本地路径没问题。flatMap(_.split(“\s”))这步是“拍平”。一行文本按空白字符切分成多个单词flatMap会把切分后的数组里的每个元素都变成RDD里的一个独立元素。如果用map结果就是RDD[Array[String]]每个元素是一个数组后续处理起来就麻烦了。这里用\s而不是空格是为了兼容多个连续空格和制表符的情况。map(word (word, 1))把每个单词映射成键值对值是1。这是MapReduce思想的典型体现——先给每个出现的词打上计数1的标记。reduceByKey(_ _)是核心聚合操作。它会把相同key的value按照传入的函数进行合并。这里就是相同单词的计数相加。reduceByKey和groupByKey的区别是个老生常谈的话题简单说reduceByKey会在map端做局部聚合combiner再shuffle到reduce端而groupByKey会把所有相同key的value直接shuffle过去再聚合。数据量大时reduceByKey的网络传输量小得多性能优势非常明显。所以能用reduceByKey就别用groupByKey。sortBy(_._2, ascending false)按词频降序排列。注意sortBy会触发一次shuffle因为要全局排序。如果数据量特别大全局排序的开销不小后面讲TopKey的时候会提到怎么避免全排序。3.2 运行结果与输出解读假设words.txt内容是这样的hello world hello spark spark is fast hello world of spark跑完程序控制台会输出(hello,3) (spark,3) (world,2) (of,1) (is,1) (fast,1)saveAsTextFile会在output/wordcount目录下生成part-00000之类的文件。注意这个目录不能已经存在否则会报FileAlreadyExistsException。这是新手最常踩的坑之一我每次跑之前都会先手动删一下output目录或者代码里加个删除逻辑。提示saveAsTextFile输出的目录里每个partition对应一个part文件文件数量等于最终RDD的partition数。如果结果集很小但part文件很多说明partition数没调好可以用coalesce或repartition合并一下再输出。3.3 几个容易踩的坑与优化点第一个坑是空行和空字符串。如果文件里有空行split之后会产生空字符串最终统计结果里会出现一个key为空的记录。处理办法是在flatMap之后加一个filter(_.nonEmpty)。第二个坑是大小写。实际文本里“Hello”和“hello”应该算同一个词但默认是区分大小写的。加一个map(_.toLowerCase)就能解决。第三个坑是标点符号。比如“hello,”和“hello”会被当成两个不同的词。简单的处理是用正则把非字母字符替换掉比如map(_.replaceAll(“[^a-zA-Z]”, “”))再filter掉空串。优化方面如果输入文件很大可以在textFile里指定第二个参数来控制partition数比如sc.textFile(path, 8)这样并行度更高。但也不是越多越好partition太多会导致任务调度开销增大一般建议每个partition处理128MB左右的数据。4. TopKey实现在WordCount基础上做排序取TopN4.1 需求分析与方案选型TopKey的需求很明确在词频统计的基础上取出出现次数最多的前N个词。看起来只是在WordCount后面加个排序和take(N)但实际写的时候有好几种方案各有优劣。最直观的方案是先reduceByKey统计词频再sortBy降序最后take(N)。这个方案逻辑最简单但sortBy是全局排序会把所有数据shuffle到一个reduce端做全排序数据量大时性能很差。而且我们只需要前N个全排序是浪费。第二种方案是用takeOrdered(N)(排序规则)。takeOrdered是Action操作它会在每个partition内部先维护一个大小为N的优先队列然后只把每个partition的TopN结果汇总到Driver端再做最终排序。这样网络传输量大大减少性能比全排序好很多。缺点是返回的是Array不是RDD如果后续还要对结果做分布式操作就不方便。第三种方案是用top(N)(隐式排序)和takeOrdered类似但top是取最大的N个语义上更直观。不过top需要元素有隐式Ordering对于(key, count)这种元组默认是按key排序的要按count排序需要自定义Ordering。我一般推荐takeOrdered方案代码简洁性能也好。如果非要用RDD做后续处理那就用sortBy加filter或者zipWithIndex的方式但那种写法复杂且性能一般。4.2 代码实现与关键算子解析package com.example.spark import org.apache.spark.sql.SparkSession object TopKey { def main(args: Array[String]): Unit { val spark SparkSession.builder() .appName(TopKeyDemo) .master(local[*]) .getOrCreate() val sc spark.sparkContext sc.setLogLevel(WARN) val inputPath src/main/resources/words.txt val topN 5 val wordCounts sc.textFile(inputPath) .flatMap(_.split(\\s)) .filter(_.nonEmpty) .map(_.toLowerCase.replaceAll([^a-z], )) .filter(_.nonEmpty) .map(word (word, 1)) .reduceByKey(_ _) val topKeys wordCounts.takeOrdered(topN)(Ordering.by[(String, Int), Int](-_._2)) println(sTop $topN words:) topKeys.foreach { case (word, count) println(s$word: $count) } spark.stop() } }重点看takeOrdered那行。Ordering.by (String, Int), Int的意思是对于(String, Int)类型的元素按照第二个字段也就是count的相反数来排序。因为takeOrdered默认取最小的N个我们对count取负最小的负数就对应最大的正数这样取出来的就是count最大的N个。这个技巧很常用比自定义一个完整的Ordering子类简洁得多。如果你觉得取负不够直观也可以这样写implicit val ordering: Ordering[(String, Int)] Ordering.by(_._2).reverse val topKeys wordCounts.takeOrdered(topN)效果一样但可读性更好一些。不过要注意隐式值的作用域别让它影响到其他地方的排序逻辑。4.3 输出格式化与结果验证跑出来的结果大概是Top 5 words: spark: 3 hello: 3 world: 2 of: 1 is: 1这里有个细节当count相同时takeOrdered的顺序是不确定的因为Ordering只定义了按count比较。如果业务要求count相同时按字母序排列需要在Ordering里加二级排序Ordering.by[(String, Int), (Int, String)](x (-x._2, x._1))这样先按count降序count相同再按单词升序。实际业务里这种二级排序的需求很常见比如排行榜里分数相同按提交时间排序。验证结果是否正确可以拿WordCount的输出对比一下TopN应该就是排序后的前N条。我一般会写个简单的断言或者手动核对一下确保没有漏掉数据。5. 性能调优与常见问题排查5.1 数据倾斜那些让人头疼的“大key”数据倾斜是Spark作业最常见的性能杀手。在WordCount场景里如果某个词出现的频率极高比如日志里的“ERROR”或者停用词“the”reduceByKey时这个key对应的partition就会特别大其他partition早就跑完了它还在慢慢磨。表现就是大部分task都完成了就剩几个task卡在99%不动。解决办法有几个。加盐打散是最常用的给key加一个随机前缀先做一次局部聚合再去掉前缀做全局聚合。代码大概是这样val salted wordCounts.map { case (word, count) val salt scala.util.Random.nextInt(10) (s$salt#$word, count) }.reduceByKey(_ _) .map { case (saltedWord, count) val word saltedWord.split(#, 2)(1) (word, count) }.reduceByKey(_ _)这样第一次reduceByKey时同一个词被分散到10个不同的key上并行度提高了。第二次再去掉盐做最终聚合此时数据量已经小了很多倾斜的影响就小了。另一个办法是提高reduce并行度在reduceByKey里指定partition数比如reduceByKey(_ _, 20)。但这只能缓解不能根治严重的倾斜。5.2 序列化与内存配置Spark默认用Java序列化性能一般。换成Kryo序列化能明显提升性能尤其是shuffle数据量大的时候spark.conf.set(spark.serializer, org.apache.spark.serializer.KryoSerializer) spark.conf.set(spark.kryo.registrationRequired, false)内存方面如果遇到OOM先看是Driver端还是Executor端。Driver端OOM通常是collect()把太多数据拉到本地了改成take(N)或者分批次collect。Executor端OOM一般是partition数据量太大可以增加partition数或者调大executor内存。我一般会把spark.executor.memory设成4g到8g具体看数据量。5.3 常见报错速查表报错信息可能原因解决办法FileAlreadyExistsException输出目录已存在删除输出目录或换路径NoSuchMethodError依赖版本不匹配检查Spark和Scala版本是否对齐OutOfMemoryError: Java heap space内存不足调大executor内存或增加partition数Task not serializable闭包里引用了不可序列化的对象把对象定义在map内部或用transient标记Shuffle fetch failedExecutor挂了或网络问题调大spark.network.timeout检查集群状态这张表里的问题我基本都遇到过尤其是Task not serializable新手很容易在闭包里引用外部的一个类实例结果序列化失败。解决办法就是把需要的字段单独取出来别整个对象往闭包里塞。6. 从案例到实战还能怎么扩展WordCount和TopKey这两个案例虽然基础但它们是很多实际需求的骨架。比如统计每个用户的访问次数TopN只需要把单词换成用户ID逻辑完全一样。再比如分析日志里出现频率最高的异常类型也是同样的套路只是数据源从文本文件换成了日志文件切分规则换成正则匹配异常堆栈。如果想进一步练手可以试试这几个方向用DataFrame API重写一遍对比RDD和DataFrame的性能差异接入流式数据用Structured Streaming做实时词频统计把结果写入数据库用foreachPartition批量插入MySQL或者Redis。这些扩展都能帮你把Spark的核心能力串起来。我个人在实际操作中的体会是别小看基础案例。很多人急着学机器学习、图计算结果连reduceByKey和groupByKey的区别都说不清楚遇到性能问题就抓瞎。把WordCount和TopKey写熟写到不用查文档就能默写出来的程度后面学更复杂的东西会顺很多。另外每次写完代码别只看结果对不对多想想有没有更优的写法多看看Spark UI里的stage划分和shuffle数据量这些习惯比多写几个案例更有价值。