简介面向大数据处理学习者与开发者这份实战代码包围绕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 和 Spark2.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.version3.3.6/hadoop.version spark.version3.4.2/spark.version scala.binary.version2.12/scala.binary.version maven.compiler.source8/maven.compiler.source maven.compiler.target8/maven.compiler.target /properties dependencies !-- 编译期要用运行期由集群提供所以是 provided -- dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version${hadoop.version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-core_${scala.binary.version}/artifactId version${spark.version}/version scopeprovided/scope exclusions !-- 排除后由集群统一版本避免 guava 打架 -- exclusion groupIdcom.google.guava/groupId artifactIdguava/artifactId /exclusion /exclusions /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-sql_${scala.binary.version}/artifactId version${spark.version}/version scopeprovided/scope /dependency /dependenciesscope填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就够别用 assemblyshade 能自动处理META-INF/services下的 SPI 文件合并少了这一步某些序列化器或者文件系统的实现类在集群上会加载不到。提示如果本地要跑单元测试把provided改成compile再跑测完记得改回来别把带 Spark 的胖包提交上去。2.3 数据契约输入样例、分隔符与输出命名示例工程里所有算法共用一套“数据契约”这套约定决定了你换自己数据时要不要动代码约定项取值改这里会影响文本编码UTF-8中文乱码读出来是方块列分隔符单空格或\tsplit正则切错会数组越界文件布局目录目录下多个 part 文件用/*通配还是读整个目录输出目录必须不存在重跑必须先删输出文件名part-r-00000/part-00000下游脚本匹配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 MapperLongWritable, 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); } } }MapperLongWritable, Text, Text, IntWritable四个泛型分别是输入 key行在文件里的字节偏移、输入 value这一行的文本、输出 key、输出 value。偏移量基本用不上用LongWritable只是为了签名对齐。关键点是outKey声明在类字段上循环里只调set。Hadoop 的序列化框架在context.write之后会把对象写出去但如果你把同一个Text对象反复塞进一个 List 再统一写出最后所有元素都会是最后一次set的值这个坑新手踩得最多。Reducer 这边public class WordCountReducer extends ReducerText, IntWritable, Text, IntWritable { private final IntWritable result new IntWritable(); Override protected void reduce(Text key, IterableIntWritable 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); } }IterableIntWritable 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 WritableComparableYearTempKey { 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 PartitionerYearTempKey, 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, IterableText values, Context context) throws IOException, InterruptedException { // values 已按温度倒序第一条就是最高温 IteratorText 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 上直接抛NullPointerExceptionjava.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-indexlocal[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_JARtarget/bigdata-algorithm-src-1.0.jar MAIN_CLASScom.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 20set -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 提交后报 ClassNotFoundExceptionjar 里没带主类现象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.n50 可覆盖Configuration的优先级是命令行-D高于资源文件这样同一份 jar 在不同环境用不同参数不用重新打包。第二步换数据源。示例读的是纯文本业务数据大概率是带分隔符的 CSV 或者需要压缩。别急着写自定义InputFormat先用TextInputFormat加split预处理能覆盖八成场景。只有遇到“一行拆成多行”、“需要按块读二进制”时才动手写InputFormat那个东西一旦写错分片边界会把记录切坏排查成本极高。第三步改算法主体。改之前先确认新算子的计算模型和示例对不对得上。对照表大概是这样业务需求可复用的示例要改的地方按维度去重取最新二次排序组合键换成维度, 时间戳比较器方向改倒序多表关联补字段reduce 端 join / broadcast join标签体系和小表裁剪逻辑分组 Top-NTop-N用TreeMap维护堆别全排序用户分群K-Means距离函数和初始中心点选择策略改算法时有个原则先别动分区和分组逻辑只改 mapper 里的解析和 reducer 里的聚合。等结果对上了再考虑加 combiner 或者换分区器优化性能。两件事一起改出了错根本分不清是哪边的问题。第四步验证。除了前面说的 shell 对拍加一层单元测试更稳。MapReduce 有MRUnitSpark 有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 吞了。从那以后我每次改完解析逻辑都强制走一遍“小样本对拍加计数器断言”两步都过了才敢往集群上提交。希望帮到你。本文还有配套的精品资源点击获取