简介：面向大数据处理学习者与开发者，这份实战代码包围绕Hadoop与Spark的数据算法实现展开，适合希望从MapReduce编程模型、RDD与DataFrame API入手，理解分布式存储、并行计算及机器学习流程的初中级读者。压缩包共876个文件，约204.27MB，以360个Java源文件和242个Jar依赖为主体，辅以34个Scala源码、31个Shell脚本、63个Markdown说明以及CSV、Parquet、PDF等数据与文档，覆盖代码实现、运行脚本、实验数据和结果文件。已有688人学习，说明其在课程实验与项目练习中具有一定参考价值。内容涉及词频统计、数据清洗转换、聚合分析，以及Spark SQL、Streaming、MLlib中的分类、回归、聚类等任务，可直接结合真实数据集运行并观察输出。目录按主题组织，便于检索源码、复现实验和积累调优思路，适合作为大数据课程配套练习与工程入门的参考资料。

1. 先把一个词频统计跑出结果，再谈集群

前阵子帮一位做离线报表的开发者看一份 Hadoop/Spark 数据算法示例源码，压缩包解开，注释齐全，目录也规整，可他把 WordCount 往集群上一提交，直接卡在ClassNotFoundException。这类源码包的价值并不在“里面塞了多少个例子”，而在于它能不能先在一个 JVM 进程里，把 MapReduce 的 key 排序逻辑和 Spark 的 shuffle 逻辑跑对，再谈搬到集群上。这份数据算法源码按 MapReduce 与 Spark 双实现整理了词频统计、倒排索引、二次排序、多表连接、Top-N、K-Means 这几类经典算子，配了小样本输入和提交脚本，适合正在啃分布式计算、需要照着改算子的人。下面按“它怎么组织、怎么跑起来、怎么改成自己的”这个顺序往下拆。

2. 工程结构与技术选型：一套源码怎么同时实现 MapReduce 和 Spark

2.1 目录布局：mr 与 spark 两棵树共用一套数据

拿到包先别急着mvn package，花两分钟把目录看一遍，后面改代码能省很多来回。这类示例工程一般会把“算法”和“引擎”分开摆：mr下每个算法一个子包，spark下按 RDD 和 SQL 再分一层，公共的 HDFS 读写、计时、配置加载抽到common。

bigdata-algorithm-src/ ├── pom.xml ├── data/ # 小样本输入，能塞进内存，本地调试全靠它 │ ├── wordcount/ │ ├── inverted-index/ │ ├── secondary-sort/ │ ├── join/ │ └── kmeans/ ├── src/main/java/com/example/bigdata/ │ ├── common/ # HdfsUtil、Timer、ConfLoader │ ├── mr/ # MapReduce 实现 │ │ ├── wordcount/ # Mapper / Reducer / Driver │ │ ├── invertedindex/ │ │ ├── secondarysort/ # 组合键 + 分区器 + 分组比较器 │ │ ├── join/ │ │ └── topn/ │ └── spark/ │ ├── rdd/ # RDD 算子版 │ └── sql/ # DataFrame / Dataset 版 ├── src/main/resources/ │ ├── log4j.properties │ └── application.conf └── scripts/ ├── run-local.sh └── submit-cluster.sh

这个布局有两个实际好处。第一，同一个算法在mr和spark下各有一份实现，输入文件复用data/里那一份，你改完 MapReduce 版，能立刻用 Spark 版对拍结果，谁对谁错一目了然。第二，common单独抽出来，意味着 HDFS 路径拼接、临时目录清理这些杂活只有一处，不至于改一个算法要动五个文件。

我一般拿到这种包，第一件事是在data/里挑最小的那个文件，用wc -l看一眼行数。行数在几千以内，本地单进程跑起来就是几秒钟的事；如果示例数据动辄上百兆，那说明作者默认你已经在集群上，本地调试得自己切一份更小的。

2.2 依赖与打包：pom.xml 里几个必须给的 scope

这类工程最容易出错的地方在pom.xml，不在代码。核心原则一句话：Hadoop 和 Spark 的依赖必须打成provided，否则你打出来的 fat jar 会和集群自带的 jar 冲突，症状是各种NoSuchMethodError、LinkageError，排查起来非常费力。

<properties> <hadoop.version>3.3.6</hadoop.version> <spark.version>3.4.2</spark.version> <scala.binary.version>2.12</scala.binary.version> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> </properties> <dependencies> <!-- 编译期要用，运行期由集群提供，所以是 provided --> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_${scala.binary.version}</artifactId> <version>${spark.version}</version> <scope>provided</scope> <exclusions> <!-- 排除后由集群统一版本，避免 guava 打架 --> <exclusion> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_${scala.binary.version}</artifactId> <version>${spark.version}</version> <scope>provided</scope> </dependency> </dependencies>

scope填provided的含义是：编译和测试阶段在 classpath 上，打包时不进 jar。这样hadoop jar和spark-submit提交上去，类由集群的 lib 目录提供，版本天然对齐。

spark-core单独排除guava是老问题了。Spark 内部依赖某个 guava 版本，Hadoop 生态里另一批组件依赖另一个版本，两边一起进 jar，运行时先加载到哪个看运气，报出来的错还经常是莫名其妙的NoSuchMethodError: com.google.common.base.Preconditions.checkArgument。我一般会统一在dependencyManagement里锁一个版本，而不是每个依赖挨个排除。

打包插件用maven-shade-plugin就够，别用 assembly，shade 能自动处理META-INF/services下的 SPI 文件合并，少了这一步，某些序列化器或者文件系统的实现类在集群上会加载不到。

提示：如果本地要跑单元测试，把provided改成compile再跑，测完记得改回来，别把带 Spark 的胖包提交上去。

2.3 数据契约：输入样例、分隔符与输出命名

示例工程里所有算法共用一套“数据契约”，这套约定决定了你换自己数据时要不要动代码：

MapReduce 侧读的是目录不是单文件，FileInputFormat.addInputPath传目录时会自动把目录下所有非_、非.开头的文件当成输入分片。Spark 侧sc.textFile传目录也一样，但传单个文件时要注意它会按行切分，跨文件的行不会混。

输出这一侧，MapReduce 的FileOutputFormat有个硬性行为：目标目录存在就直接抛FileAlreadyExistsException，不做覆盖。这是保护机制，不是 bug，后面排查章节会专门说怎么绕。

3. 核心算法拆解：一个 key 从 map 到 reduce 的完整链路

3.1 MapReduce 三段式：Mapper / Reducer / Driver

先看最小闭环。WordCount 是所有分布式教程的第一课，但这份源码里它的写法有几个细节值得抄：输出对象复用、空行过滤、combiner 复用 reducer。

package com.example.bigdata.mr.wordcount; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; import java.util.StringTokenizer; public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> { // 复用对象，避免每条记录 new 一次，map 阶段 QPS 高的时候这很关键 private final Text outKey = new Text(); private static final IntWritable ONE = new IntWritable(1); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); if (line.trim().isEmpty()) { return; // 空行不进 reduce，减少无意义的 shuffle } // 中文标点也要切，否则「数据，算法」会被算成一个词 StringTokenizer itr = new StringTokenizer(line, " \t,.;:\"'\uFF0C\u3002\uFF1B\uFF1A"); while (itr.hasMoreTokens()) { outKey.set(itr.nextToken().toLowerCase()); context.write(outKey, ONE); } } }

Mapper<LongWritable, Text, Text, IntWritable>四个泛型分别是输入 key（行在文件里的字节偏移）、输入 value（这一行的文本）、输出 key、输出 value。偏移量基本用不上，用LongWritable只是为了签名对齐。

关键点是outKey声明在类字段上，循环里只调set。Hadoop 的序列化框架在context.write之后会把对象写出去，但如果你把同一个Text对象反复塞进一个 List 再统一写出，最后所有元素都会是最后一次set的值，这个坑新手踩得最多。

Reducer 这边：

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private final IntWritable result = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; // values 是迭代器不是集合，只能遍历一次，别想着先 size() 再求和 for (IntWritable v : values) { sum += v.get(); } result.set(sum); context.write(key, result); } }

Iterable<IntWritable> values是最反直觉的地方：它不是List，底层是框架给你的一个复用迭代器，遍历完就没了。想拿总数只能累加，不能Collection化。数据量大的时候把 values 全塞进ArrayList会直接把 reduce 容器打爆，这是内存溢出最常见的原因之一。

Driver 这层是粘合剂：

public class WordCountDriver { public static void main(String[] args) throws Exception { if (args.length < 2) { System.err.println("用法: WordCountDriver <input> <output>"); System.exit(2); } Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordCountDriver.class); // 集群上找主类的依据，别漏 job.setMapperClass(WordCountMapper.class); job.setCombinerClass(WordCountReducer.class); // 逻辑满足交换律结合律才能这么复用 job.setReducerClass(WordCountReducer.class); job.setOutputKeyClass(Text.class); // 注意：这是最终输出类型 job.setOutputValueClass(IntWritable.class); // map 输出类型和最终输出不一致时，这两个必须单独设 // job.setMapOutputKeyClass(Text.class); // job.setMapOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }

setJarByClass决定了 YARN 从哪个 jar 里找你的类，漏了它最常见的症状是提交成功、然后一直ACCEPTED之后失败，日志里说找不到 mapper。setCombinerClass复用 reducer 是有前提的：合并操作必须满足交换律和结合律。求和、求最大值可以，求平均值就不行——平均值会被合并两次，结果偏小。这条血泪经验我写在了不知道多少个项目的注释里。

3.2 二次排序：组合键、分区器与分组比较器的三件套

二次排序是这份源码里最能体现 MapReduce 设计功力的部分。需求是：按“年份”分组，组内按“温度”倒序，取每年最高温。要让它正确工作，得三个组件配合，少一个结果就是乱的。

先定义组合键：

public class YearTempKey implements WritableComparable<YearTempKey> { private final Text year = new Text(); private final IntWritable temp = new IntWritable(); public YearTempKey() {} // 反序列化必须要无参构造 public void set(String y, int t) { year.set(y); temp.set(t); } @Override public void write(DataOutput out) throws IOException { year.write(out); temp.write(out); } @Override public void readFields(DataInput in) throws IOException { year.readFields(in); temp.readFields(in); } @Override public int compareTo(YearTempKey o) { int cmp = year.compareTo(o.year); if (cmp != 0) { return cmp; } // 温度倒序：同一年里最大值排最前，reduce 里取第一条就是答案 return -Integer.compare(temp.get(), o.temp.get()); } @Override public int hashCode() { return year.hashCode(); // 只按年份散列，保证同一年进同一个分区 } @Override public boolean equals(Object obj) { if (!(obj instanceof YearTempKey)) return false; return year.equals(((YearTempKey) obj).year); } }

compareTo决定排序，hashCode决定分区。这两个必须自洽：compareTo里年份相同温度不同算“不相等”，但hashCode/equals只认年份，这样同年的记录一定落到同一个 reduce 分区里，同时组内又是按温度排好序的。如果hashCode把温度也算进去，同一年会被散到不同 reducer，最终输出会出现重复年份，这就是很多人二次排序“结果看着对但多了几行”的根因。

第二个组件是分区器：

public class YearPartitioner extends Partitioner<YearTempKey, Text> { @Override public int getPartition(YearTempKey key, Text value, int numPartitions) { return Math.abs(key.getYear().hashCode() * 127) % numPartitions; } }

第三个是分组比较器，它只比年份：

public class YearGroupingComparator extends WritableComparator { protected YearGroupingComparator() { super(YearTempKey.class, true); // true 表示让框架直接 new 实例，走 readFields } @Override public int compare(WritableComparable a, WritableComparable b) { YearTempKey k1 = (YearTempKey) a; YearTempKey k2 = (YearTempKey) b; // 只比年份：同年的所有温度归到一次 reduce 调用里 return k1.getYear().compareTo(k2.getYear()); } }

三个组件串起来看就清楚了：分区器把同年数据送到同一个 reducer，排序比较器让同年数据按温度倒序排列，分组比较器把“同年不同温”的记录合并成一次reduce()调用。于是 reducer 里只要取values的第一个元素，就是该年最高温。

@Override protected void reduce(YearTempKey key, Iterable<Text> values, Context context) throws IOException, InterruptedException { // values 已按温度倒序，第一条就是最高温 Iterator<Text> it = values.iterator(); if (it.hasNext()) { context.write(new Text(key.getYear()), new Text(it.next())); } }

这套三段式在 MapReduce 里非常通用：Top-N、去重取最新一条、按会话切分，都是它的变体。看懂这一处，源码里其他几个算法模块基本能举一反三。

3.3 Spark 侧的两套写法：RDD 与 DataFrame 怎么选

同一份倒排索引，源码给了两个版本。RDD 版长这样：

val lines = sc.textFile(input) val pairs = lines.flatMap { line => val parts = line.trim.split("\\s+") if (parts.length < 2) Seq.empty else { val docId = parts.head // 第一列是文档 id parts.tail.distinct.map(w => (w, docId)) // 同一文档内先去重，减少 shuffle 数据量 } } pairs.groupByKey() .mapValues(_.toSet.mkString(",")) // 拼成 a.txt,b.txt .sortByKey(numPartitions = 1) // 排序输出，全局有序必须单分区 .saveAsTextFile(output)

flatMap负责一对多展开，groupByKey把同一个词的文档聚到一起。这里有个性能隐患：groupByKey会把同一个 key 的所有 value 全拉到内存里，词出现在几百万个文档中时，Executor 直接 OOM。数据量上去以后我一般换成aggregateByKey，用mutable.Set增量合并，内存占用可控。

sortByKey(1)里的 1 是分区数，全序输出只能靠单分区，代价是所有数据往一个 task 上汇。如果只是“每个分区内有序”，分区数就按集群核数给。

DataFrame 版走了另一条路：

import org.apache.spark.sql.functions._ val df = spark.read.text(input) val docWords = df.map { r => val arr = r.getString(0).trim.split("\\s+") (arr.head, arr.tail.distinct) }.toDF("doc_id", "words") docWords .select($"doc_id", explode($"words").as("word")) .groupBy("word") .agg(collect_set("doc_id").as("docs")) .select($"word", concat_ws(",", $"docs").as("doc_list")) .orderBy("word") .write.text(output)

DataFrame 版本的好处是 Catalyst 会帮你做谓词下推、列裁剪，groupBy走的也是 Tungsten 的聚合路径，比 RDD 的groupByKey稳。代价是explode之后行数膨胀，collect_set本身又是一个全量聚合，内存压力还是在那。

我的选择标准很朴素：输入是非结构化文本、逻辑要写复杂控制流，用 RDD；输入有明确 schema、只做过滤聚合排序，用 DataFrame。两套代码在同一个工程里并存，本身就是让你对着比较的。

3.4 多表连接与数据倾斜：reduce 端 join 与 broadcast join

源码里 join 模块给了两种实现，对应两种数据规模。

reduce 端 join 的思路是给每条记录打标签，按 join key 分区，reducer 里做笛卡尔积：

// Mapper：根据来源打标签，输出 <joinKey, "A:字段"> 或 <joinKey, "B:字段"> String line = value.toString(); String[] f = line.split(",", 3); if (f.length < 3) return; String tag = fileName.contains("order") ? "A" : "B"; context.write(new Text(f[0]), new Text(tag + ":" + f[1] + "," + f[2]));

Reducer 里把两份数据分开暂存，再两两组合输出。这个方案能处理大表 join 大表，但 shuffle 数据量等于两张表之和，网络开销大。

另一种是 broadcast join，适合一张表小到能放进内存：

// 小表收集到 Driver，广播到每个 Executor val dict = sc.textFile(smallPath) .map { line => val a = line.split(",", 2) (a(0), a(1)) } .collectAsMap() val dictBC = sc.broadcast(dict) // 大表逐条查广播变量，全程零 shuffle val joined = sc.textFile(bigPath).flatMap { line => val a = line.split(",", 3) dictBC.value.get(a(1)).map(city => (a(0), city, a(2))) }

collectAsMap会把小表全拉到 Driver，所以小表必须真的小——几十万行以内比较安全，上百万行就有 Driver OOM 的风险。sc.broadcast把它分发到每个 Executor 的块管理器里，大表的 task 直接本地查表，省掉了整个 shuffle 阶段。

判断用哪种很简单：小表能进内存就广播，进不去就走 reduce 端 join。中间态（几十万到几百万行）是最难受的，我一般会先对小表做一次聚合裁剪，只保留 join 用得到的列，压缩到能广播的体量。

4. 三级验证跑法：本地单进程、伪分布式与集群提交

4.1 本地跑通：不装集群先把逻辑跑对

搭集群之前，先在本地把逻辑跑对。Hadoop 的LocalJobRunner是单 JVM 模拟 map 和 reduce，不需要任何守护进程。

# 1. 编译打包，跳过测试先求快 mvn -q clean package -DskipTests # 2. 用示例数据跑 WordCount，输入输出都走本地文件系统 hadoop jar target/bigdata-algorithm-src-1.0.jar \ com.example.bigdata.mr.wordcount.WordCountDriver \ data/wordcount/in \ out/wordcount

第一次跑大概率会碰到两个问题。一是命令行里hadoop找不到，得先有HADOOP_HOME并把它加进PATH。二是在 Windows 上直接抛NullPointerException，java.lang.NullPointerException: null at java.io.File.<init>，原因是缺 Hadoop 的本地库支持文件，需要额外配置HADOOP_HOME并把bin目录塞进去（Windows 上这个问题会一直伴随到你换 Mac 或者容器）。

Spark 侧本地跑更省事，local[*]用满本机核数：

spark-submit \ --master "local[2]" \ --class com.example.bigdata.spark.rdd.InvertedIndexApp \ target/bigdata-algorithm-src-1.0.jar \ data/inverted-index/in \ out/inverted-index

local[2]里的 2 是并发 task 数，本地调试给 2 到 4 就够，给太多反而因为线程切换看不出真实瓶颈。如果想验证分区逻辑，把它调成 1 和 8 各跑一次，结果应该完全一致——不一致说明代码里依赖了分区顺序，那是隐藏 bug。

4.2 集群提交：脚本里那几个变量怎么填

本地跑通之后，把输入先推上去：

# 建目录、传数据 hdfs dfs -mkdir -p /data/wordcount/in hdfs dfs -put -f data/wordcount/in/* /data/wordcount/in/ hdfs dfs -ls /data/wordcount/in

提交脚本我一般写成带时间戳的，避免反复重跑撞上“输出目录已存在”：

#!/usr/bin/env bash set -euo pipefail APP_JAR=target/bigdata-algorithm-src-1.0.jar MAIN_CLASS=com.example.bigdata.mr.secondarysort.SecondarySortDriver HDFS_IN=/data/secondary-sort/in # 时间戳做后缀，每次都是新目录，不用手动删 HDFS_OUT=/data/secondary-sort/out-$(date +%Y%m%d%H%M%S) # 输入不存在提前失败，别让作业白跑一轮 hdfs dfs -test -e "$HDFS_IN" || { echo "输入路径不存在: $HDFS_IN"; exit 1; } hadoop jar "$APP_JAR" "$MAIN_CLASS" "$HDFS_IN" "$HDFS_OUT" # 立刻回读前 20 行，确认有输出 hdfs dfs -cat "$HDFS_OUT"/part-r-* | head -n 20

set -euo pipefail三件套别省，尤其-e，作业失败脚本要立刻停，不然下一步hdfs dfs -cat会在不存在的目录上报一串更迷惑的错。

Spark 提交则要注意--driver-memory和--executor-memory的分配。广播变量占的是 Executor 内存，小表几百万行的时候，--executor-memory 2g很可能不够，加到 4g 再试。

4.3 结果对拍：拿 shell 一行命令验算

分布式作业最怕“跑完了、有输出、结果是错的”。小样本数据完全可以手动验算，用一行命令对拍：

# 用 shell 算出标准答案：拆词、排序、去重计数 cat data/wordcount/in/* | tr -s ' \t' '\n' \ | tr -d '.,;:' | tr 'A-Z' 'a-z' | grep -v '^$' \ | sort | uniq -c | awk '{print $2"\t"$1}' | sort > /tmp/expect.txt # 拉回作业输出 hdfs dfs -getmerge /data/wordcount/out-* /tmp/actual.txt diff <(sort /tmp/expect.txt) <(sort /tmp/actual.txt) && echo "结果一致"

getmerge会把目录下所有 part 文件合并成一个本地文件，比逐个cat方便。diff没输出就说明一致，有差异先看数量级——差一倍往往是 combiner 用错，差几行往往是分隔符或编码问题。

除了对拍，MapReduce 的计数器也是免费的质量指标。作业结束日志里的Map input records、Reduce input records、Combine input records三个数能反推很多东西：combiner 生效了的话，Combine input records应该大于Reduce input records。

5. 排查手册：5 个把示例代码跑挂的常见问题

5.1 提交后报 ClassNotFoundException：jar 里没带主类

现象：hadoop jar提交后作业状态先ACCEPTED再FAILED，日志里java.lang.ClassNotFoundException: com.example.bigdata.mr.wordcount.WordCountMapper。

原因：Job.setJarByClass传的是 Driver 类，但如果 shade 打包时主类被 relocate 或者 jar 里根本没有这个类，YARN 在容器里就加载不到。另一种情况是打的是-sources.jar或者original-xxx.jar，那个包里只有依赖没有你的类。

解决：先unzip -l target/xxx.jar | grep WordCountMapper确认类在不在包里。不在就检查 shade 插件的includes配置，是不是把com/example/**排掉了。如果在包里还报错，那是提交时指定了错误的 jar 路径，检查脚本里的APP_JAR变量是不是相对路径、在别的目录执行时失效。

5.2 本地跑 MR 报 NullPointerException：缺 HADOOP_HOME

现象：IDE 里直接 run 主类，Configuration一构造就抛NullPointerException，堆栈指向org.apache.hadoop.util.Shell。

原因：Hadoop 在初始化时会探测本地平台，Windows 下需要HADOOP_HOME指向一个带bin目录的解压路径，里面要有winutils这类本地支持文件。缺了它，构造FileSystem时拿到的路径是 null，于是空指针。

解决：配置HADOOP_HOME环境变量指向本地目录，把%HADOOP_HOME%\bin加进PATH，再把对应的本地库文件放进bin。改完重启 IDE，环境变量不重启不生效。长期方案是用 WSL 或者容器跑，能少掉一大半这类平台问题。

注意：这个报错和业务代码无关，别去翻 Mapper 的实现，先确认环境。

5.3 二次排序输出顺序乱：分组比较器和 compareTo 对不上

现象：二次排序结果里同一年份出现多行，或者组内温度不是倒序，甚至每年只有一条但取的不是最高温。

原因：三个组件里有一个没生效。最常见的是job.setGroupingComparatorClass忘了调，框架退化成用 key 的compareTo分组，于是“同年不同温”被当成不同的组，每次 reduce 只拿到一条记录。第二种是hashCode里把温度也算进去了，同年数据散到了不同 reducer。

解决：Driver 里三行一个都不能少——setPartitionerClass、setSortComparatorClass（默认用 key 的compareTo，可以不设）、setGroupingComparatorClass。再检查compareTo和分组比较器的方向是否一致：compareTo里温度用了负号做倒序，分组比较器只比年份，两者不能互相干扰。

5.4 reduce 卡在 99%：某个 key 的数据倾斜

现象：作业进度条长时间停在reduce 99%，只有一两个 reduce task 还在跑，其余全部完成。日志里能看到某个 task 处理字节数远高于其他。

原因：join key 或者 group key 分布极不均匀。比如按城市聚合，一线城市的记录数是小城市的几百倍，所有数据涌向同一个 reducer，其余 reducer 早早完工在等它。

解决：分三层处理。第一层，能在 map 端聚合的先聚合，用 combiner 压缩数据量。第二层，给热点 key 加随机前缀打散，做一次局部聚合再去掉前缀做全局聚合。第三层，join 场景下如果热点 key 对应的是小表，直接换成 broadcast join，从根上消除 shuffle。

// 热点 key 打散：加 0~9 的随机后缀，局部聚合后再合并 val salted = rdd.map { case (k, v) => (s"${k}_${scala.util.Random.nextInt(10)}", v) } .reduceByKey(_ + _) .map { case (k, v) => (k.split("_")(0), v) } .reduceByKey(_ + _)

两轮 reduceByKey 比一轮多一次 shuffle，但把单点压力摊到了 10 个 task 上，总体耗时反而下降。这个技巧在源码的 Top-N 模块里有现成注释，值得对着看。

5.5 重跑报 FileAlreadyExistsException：输出目录没清

现象：第二次执行同一个命令，作业启动后立刻失败，org.apache.hadoop.mapred.FileAlreadyExistsException: Output directory hdfs://.../out already exists。

原因：FileOutputFormat的行为约定就是拒绝覆盖，防止误删数据。

解决：三种做法。开发阶段在 Driver 里手动删：FileSystem.get(conf).delete(new Path(args[1]), true)，第二个参数true表示递归删。脚本阶段用时间戳后缀，每次新目录。生产环境别自动删，改成失败退出，由调度系统管理生命周期。

提示：delete的递归参数一定给true，给false时目录非空会删不掉，然后报一个和权限无关的错，浪费时间。

6. 二次开发：把示例算法改成业务算子的四步走

源码跑通只是起点，真正要用起来，得把它改成自己业务里的算子。我一般按四步走，每步都能独立验证。

第一步，抽骨架。把 Driver 里会变的三样东西参数化：输入路径、输出路径、业务参数（比如 Top-N 的 N、K-Means 的 K）。别把参数硬编码在类里，改成从args或者配置文件读：

// 参数化后的 Driver 片段：业务参数从配置读，路径从命令行读 Configuration conf = new Configuration(); conf.addResource("application.conf"); // 默认值放这里 int topN = conf.getInt("app.topn.n", 10); // 命令行 -Dapp.topn.n=50 可覆盖

Configuration的优先级是命令行-D高于资源文件，这样同一份 jar 在不同环境用不同参数，不用重新打包。

第二步，换数据源。示例读的是纯文本，业务数据大概率是带分隔符的 CSV 或者需要压缩。别急着写自定义InputFormat，先用TextInputFormat加split预处理，能覆盖八成场景。只有遇到“一行拆成多行”、“需要按块读二进制”时才动手写InputFormat，那个东西一旦写错，分片边界会把记录切坏，排查成本极高。

第三步，改算法主体。改之前先确认新算子的计算模型和示例对不对得上。对照表大概是这样：

改算法时有个原则：先别动分区和分组逻辑，只改 mapper 里的解析和 reducer 里的聚合。等结果对上了，再考虑加 combiner 或者换分区器优化性能。两件事一起改，出了错根本分不清是哪边的问题。

第四步，验证。除了前面说的 shell 对拍，加一层单元测试更稳。MapReduce 有MRUnit，Spark 有SparkSession的本地模式，都能在小数据上跑断言。我习惯在 reducer 里加计数器，统计异常记录数：

// 解析失败的行单独计数，跑完看计数器就知道脏数据比例 try { String[] f = line.split(",", 3); if (f.length < 3) throw new IllegalArgumentException("字段不足"); context.write(new Text(f[0]), new Text(f[1])); } catch (Exception e) { context.getCounter("APP", "BAD_RECORD").increment(1); }

作业跑完，BAD_RECORD的数值直接告诉你数据质量怎么样，比事后去翻日志快得多。这个习惯是从一次线上事故之后养成的——那次输出结果少了三成，查了一下午才发现是某个字段为空导致split后数组长度不够，异常被 catch 吞了。从那以后我每次改完解析逻辑，都强制走一遍“小样本对拍加计数器断言”，两步都过了才敢往集群上提交。希望帮到你。

本文还有配套的精品资源，点击获取