简介：这份资源是面向大数据与Java Web初学者及课程实践者的Hadoop气象数据分析完整代码包，围绕HDFS分布式存储与MapReduce并行计算展开，覆盖气温、最低温度等气象指标的统计场景，并配套SSM框架将分析结果以网页形式展示，帮助读者打通从数据清洗、分组聚合到Web呈现的完整链路。压缩包共562个文件、约34.88MB，以js、css、png、jpg等前端与静态资源为主，辅以xml配置、jar与class编译产物、java源码、jsp页面及sql脚本，另有war包便于直接部署，目录结构清晰，适合按模块对照学习。目前已有10598人学习下载，读者可从中获得Map与Reduce阶段的键值对设计思路、气象指标聚合实现、SSM整合配置以及项目打包部署的参考，是理解Hadoop生态实际应用的一份可运行实践素材。

1. 从一堆气象站日志说起：这套 Hadoop 代码到底解决什么问题

手头有一批气象观测数据，按站点、日期、气温、气压、风速这些字段一行行堆着，单机跑个脚本处理几百万行还行，数据量再往上走，磁盘 IO 和内存就开始拖后腿。这套 Hadoop 气象数据完整版代码要解决的正是这个场景：把原始气象日志丢进 HDFS，用 MapReduce 做分布式清洗和统计，输出每天的最高气温、最低气温、平均气温这类指标。它适合两类人——正在学 MapReduce 编程模型、想找一个能跑通的完整案例的开发者，以及手上真有气象或传感器时序数据、想用 Hadoop 做批量离线分析的人。代码结构不复杂，但把输入格式、Mapper、Reducer、驱动类这条链路走全了，拿来当模板改造成自己的统计任务很顺手。

2. 项目结构与技术栈：从 HDFS 输入到统计结果落盘

2.1 整体架构与数据流

这套代码的骨架是标准的 Hadoop MapReduce 批处理链路，没有引入 Hive、Spark 这类上层框架，好处是依赖干净、逻辑透明，坏处是所有清洗和聚合都得自己写。数据流大致是这样：原始气象日志先上传到 HDFS 的输入目录，驱动类WeatherDriver配置 Job 参数并提交任务，Hadoop 按输入分片切分文件，每个分片交给一个 Mapper 实例处理，Mapper 输出<日期, 气温>这样的键值对，经过 Shuffle 阶段按日期分组排序后送到 Reducer，Reducer 对同一天的所有气温做聚合，最后结果写回 HDFS 输出目录。

这个链路里有两个关键设计点值得注意。第一是 Mapper 输出的 key 选择，选日期作为 key 意味着同一天的数据一定会落到同一个 Reducer，这是聚合正确性的前提；第二是 Combiner 的取舍，气温统计里最大值和最小值满足结合律，可以安全地用 Combiner 在 Map 端预聚合，减少网络传输量，但平均值不行，因为平均值不满足结合律，得拆成 sum 和 count 两个值分别聚合。这套代码在实现上对这两点做了区分处理，后面拆代码时会细说。

技术栈方面，核心依赖就三个：Hadoop 客户端库（提供Mapper、Reducer、Job这些基类）、JDK（代码是 Java 写的，编译需要 JDK 8 及以上）、构建工具（Maven 或直接 javac 加 classpath 都行）。运行环境需要一套可用的 Hadoop 集群，伪分布式模式就够跑通，生产环境再换成完全分布式。数据格式上，原始日志是纯文本，每行一条记录，字段之间用空格或逗号分隔，具体分隔符在 Mapper 里解析时指定。

2.2 目录结构与文件职责

代码包解压后，目录大致是这么组织的：

weather-hadoop/ ├── pom.xml # Maven 构建配置，声明 Hadoop 依赖 ├── src/main/java/com/example/weather/ │ ├── WeatherDriver.java # 驱动类，配置 Job 并提交 │ ├── WeatherMapper.java # Mapper，解析日志并输出中间键值对 │ ├── WeatherReducer.java # Reducer，按日期聚合气温 │ └── WeatherCombiner.java # Combiner，Map 端预聚合（可选） ├── src/main/resources/ │ └── log4j.properties # 日志配置，避免控制台刷屏 └── data/ └── sample-weather.txt # 样例数据，用于本地测试

这个结构是 Maven 标准布局，pom.xml里声明hadoop-client依赖，版本要和集群上跑的 Hadoop 版本对齐，否则容易出现NoSuchMethodError这类运行时问题。WeatherDriver是入口，main方法里组装 Job；WeatherMapper和WeatherReducer是核心逻辑；WeatherCombiner是可选的优化项，不配也能跑，配了能减少 Shuffle 数据量。data/sample-weather.txt是给本地调试用的，格式和真实日志一致，方便先在单机上验证逻辑再上集群。

2.3 环境准备与依赖配置

跑这套代码之前，环境得先备齐。JDK 装 8 或 11 都行，Hadoop 建议用 3.x 系列，伪分布式模式按官方文档配好core-site.xml和hdfs-site.xml，确保hdfs dfs -ls /能正常列出目录。Maven 装好后，pom.xml里加依赖：

<dependencies> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.3.6</version> <!-- 与集群 Hadoop 版本保持一致 --> </dependency> </dependencies>

版本号这里要特别注意，hadoop-client的版本必须和集群上运行的 Hadoop 版本一致，差一个小版本都可能因为 RPC 协议不兼容导致任务提交失败。如果集群是 3.3.x，客户端就用 3.3.x；集群是 2.7.x，客户端也得降到 2.7.x。这是踩过坑的地方，后面避坑章节会展开。

依赖拉下来之后，用mvn clean package打出一个 fat jar，或者用mvn dependency:copy-dependencies把依赖 jar 拷到target/dependency目录，提交任务时用-libjars参数带上。前者简单，后者灵活，看集群环境怎么方便怎么来。

3. 核心代码拆解：Mapper、Reducer 与驱动类逐段讲

3.1 Mapper：日志解析与中间键值对生成

Mapper 的职责是把一行原始日志解析成<日期, 气温>这样的键值对。气象日志的典型格式是「站点编号 日期 气温 气压 风速」，字段之间用空格分隔，气温可能是负数，解析时要处理符号。代码大致长这样：

public class WeatherMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private final Text outKey = new Text(); private final IntWritable outValue = new IntWritable(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString().trim(); // 跳过空行和表头 if (line.isEmpty() || line.startsWith("station")) { return; } String[] fields = line.split("\\s+"); // 字段数不足说明是脏数据，直接丢弃 if (fields.length < 3) { return; } String date = fields[1]; try { int temp = Integer.parseInt(fields[2]); outKey.set(date); outValue.set(temp); context.write(outKey, outValue); } catch (NumberFormatException e) { // 气温字段解析失败，记录后跳过，不中断整个任务 context.getCounter("WEATHER", "BAD_TEMP").increment(1); } } }

这段代码有几个细节值得说。输入 key 是LongWritable，表示行偏移量，实际用不上，但 MapReduce 的Mapper基类签名要求这个类型。line.split("\\s+")用正则匹配一个或多个空白字符，比单个空格更稳，因为日志里字段之间可能有多个空格或制表符。字段数校验和NumberFormatException捕获是防御性编程，真实日志里脏数据不少，直接抛异常会让整个任务失败，用计数器记录脏数据数量更合理，任务跑完看计数器就知道丢了多少条。

输出 key 选日期字符串，value 选气温整数。这里用IntWritable而不是Text，是因为后续聚合要做数值比较，用整型省去反复解析的开销。outKey和outValue声明为成员变量并在map方法里复用，是 Hadoop 的惯用写法，避免每行都创建新对象，减少 GC 压力。

3.2 Reducer：按日期聚合最高最低气温

Reducer 收到的是<日期, [气温1, 气温2, ...]>这样的输入，要对同一天的所有气温求最大值和最小值。代码：

public class WeatherReducer extends Reducer<Text, IntWritable, Text, Text> { private final Text outValue = new Text(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int max = Integer.MIN_VALUE; int min = Integer.MAX_VALUE; long sum = 0; int count = 0; for (IntWritable val : values) { int temp = val.get(); if (temp > max) max = temp; if (temp < min) min = temp; sum += temp; count++; } // 输出格式：最高温,最低温,平均温 double avg = count > 0 ? (double) sum / count : 0.0; outValue.set(max + "," + min + "," + String.format("%.1f", avg)); context.write(key, outValue); } }

reduce方法的values是一个迭代器，Hadoop 框架保证同一个 key 的所有 value 会一次性传给同一个reduce调用，所以可以在一次遍历里同时算出最大值、最小值和平均值。这里没有用 Combiner 做平均值预聚合，原因前面提过——平均值不满足结合律，Map 端算出的局部平均值再求平均会失真。如果要用 Combiner，得把输出 value 改成「sum,count,max,min」这样的复合结构，Reducer 再做二次聚合，代码会复杂一些，这套代码选择了简单直接的写法。

输出 value 用Text拼成逗号分隔的字符串，方便后续用hdfs dfs -cat直接看结果。如果下游还要接其他处理，可以改成DoubleWritable或自定义Writable，看具体需求。

3.3 驱动类：Job 配置与提交

驱动类负责把 Mapper、Reducer、输入输出路径这些组装成一个 Job 并提交给集群：

public class WeatherDriver { public static void main(String[] args) throws Exception { if (args.length != 2) { System.err.println("Usage: WeatherDriver <input> <output>"); System.exit(1); } Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "weather stats"); job.setJarByClass(WeatherDriver.class); job.setMapperClass(WeatherMapper.class); job.setReducerClass(WeatherReducer.class); job.setCombinerClass(WeatherCombiner.class); // 可选，见 3.4 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }

Job.getInstance(conf, "weather stats")创建 Job 实例，第二个参数是任务名，会在 YARN 的 Web UI 上显示，起个有意义的名字方便排查。setJarByClass告诉 Hadoop 用哪个类所在的 jar 包来分发任务代码，集群模式下这行不能少，否则 TaskTracker 找不到 Mapper 和 Reducer 的类。setMapOutputKeyClass和setMapOutputValueClass指定 Mapper 输出的类型，setOutputKeyClass和setOutputValueClass指定最终输出的类型，这两组类型不一致时（比如 Mapper 输出IntWritable但 Reducer 输出Text）必须分别设置，只设一组会报类型不匹配。

输入输出路径从命令行参数传入，FileInputFormat.addInputPath可以多次调用添加多个输入目录，FileOutputFormat.setOutputPath指定的输出目录必须不存在，否则 Job 提交时直接报错。这是 Hadoop 的保护机制，防止误覆盖已有结果，重跑任务前记得先删输出目录或者换个路径。

3.4 Combiner：Map 端预聚合的取舍

Combiner 是可选的优化组件，在 Map 端对输出做一次局部聚合，减少传到 Reducer 的数据量。对气温统计来说，最大值和最小值可以安全预聚合，平均值不行。这套代码里的WeatherCombiner如果只做 max/min 预聚合，逻辑和 Reducer 类似但输出类型要匹配 Mapper 的输出类型：

public class WeatherCombiner extends Reducer<Text, IntWritable, Text, IntWritable> { private final IntWritable outValue = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int max = Integer.MIN_VALUE; for (IntWritable val : values) { if (val.get() > max) max = val.get(); } outValue.set(max); context.write(key, outValue); } }

注意这个 Combiner 只保留了最大值，最小值丢了，所以它不能直接配给 3.2 里那个同时算 max/min/avg 的 Reducer 用——配上去结果会错。正确做法要么是 Combiner 输出复合结构（max,min,sum,count），Reducer 做二次聚合；要么干脆不配 Combiner，接受多一点网络传输。这套代码在WeatherDriver里把setCombinerClass注释掉了，就是这个原因。如果读者要加 Combiner 优化，务必确认聚合函数满足结合律，并且 Combiner 的输出类型和 Mapper 的输出类型一致，否则运行时会抛类型转换异常。

4. 编译、提交与结果验证：从本地伪分布式到集群

4.1 本地编译与样例数据试跑

先在本地把代码编译打包，确认逻辑没问题再上集群。用 Maven 的话：

# 编译打包，跳过测试加速 mvn clean package -DskipTests # 产物在 target/ 下，通常是 weather-hadoop-1.0.jar ls target/*.jar

打包完成后，准备一份样例数据放到 HDFS：

# 在 HDFS 上建输入目录 hdfs dfs -mkdir -p /user/weather/input # 上传样例数据 hdfs dfs -put data/sample-weather.txt /user/weather/input/ # 确认文件已上传 hdfs dfs -ls /user/weather/input/

样例数据不用太大，几百行就够验证逻辑，格式要和 Mapper 里的解析规则对齐——空格分隔、第二列是日期、第三列是气温。如果样例数据和真实日志格式有差异，先在本地用cat加awk看一眼字段分布，确认分隔符和列序再上传，省得跑完任务发现结果全是脏数据计数。

4.2 提交 Job 与观察运行状态

提交任务用hadoop jar命令：

hadoop jar target/weather-hadoop-1.0.jar \ com.example.weather.WeatherDriver \ /user/weather/input \ /user/weather/output

第一个参数是 jar 包路径，第二个是驱动类全限定名，后面两个是输入和输出路径。提交后控制台会打印任务进度，包括 Map 和 Reduce 的完成百分比。如果想看更详细的状态，打开 YARN 的 Web UI（默认 8088 端口），找到对应的 application，能看到每个 Task 的日志和计数器。计数器里WEATHER.BAD_TEMP就是 Mapper 里记录的气温解析失败条数，跑完看一眼这个数，如果占比很高说明数据格式和解析逻辑不匹配，得回去改 Mapper。

任务跑完后，输出目录下会有part-r-00000这样的结果文件，用hdfs dfs -cat查看：

hdfs dfs -cat /user/weather/output/part-r-00000 | head -20

每行格式是「日期 最高温,最低温,平均温」，按日期字典序排列。想验证结果对不对，挑几天用原始数据手工算一遍对比，或者用awk在本地对样例数据做同样的聚合，两边结果一致就说明逻辑没问题。

4.3 结果校验与常见输出异常

结果校验除了手工对比，还可以看几个指标：输出行数是否等于输入数据里不同日期的数量（用hdfs dfs -cat output/* | wc -l和cut -d' ' -f2 input/* | sort -u | wc -l对比），以及有没有出现Integer.MIN_VALUE或Integer.MAX_VALUE这种初始值残留——如果某天的 values 为空（理论上不会，但数据异常时可能），Reducer 里的 max/min 会保持初始值，输出就会很离谱。这种情况一般是 Mapper 输出的 key 有问题，比如日期字段解析出了空字符串。

另一个常见异常是输出目录已存在导致 Job 提交失败，报org.apache.hadoop.mapred.FileAlreadyExistsException。解决方法是换一个输出路径，或者先hdfs dfs -rm -r /user/weather/output删掉旧目录。生产环境里我一般会在驱动类里加一段逻辑，检测输出路径存在时自动加时间戳后缀，避免手动删目录的麻烦。

5. 避坑与排查：跑这套代码时最容易翻车的几个点

5.1 客户端与集群 Hadoop 版本不一致

现象：任务提交后卡在ACCEPTED状态很久，或者直接抛NoSuchMethodError、ClassNotFoundException，日志里出现 RPC 协议相关的异常。

原因：本地pom.xml里hadoop-client的版本和集群上运行的 Hadoop 版本不一致，RPC 协议或类签名对不上。Hadoop 小版本之间虽然兼容性较好，但跨大版本（比如 2.x 客户端连 3.x 集群）基本必挂。

解决：把pom.xml里的版本改成和集群完全一致，用hadoop version命令确认集群版本号，然后重新打包。如果集群是多版本共存，提交时用HADOOP_HOME指向对应版本的客户端配置目录。

5.2 输出目录已存在导致任务直接失败

现象：Job 提交后立刻失败，异常信息里有FileAlreadyExistsException，提示输出路径已存在。

原因：Hadoop 默认不允许覆盖已存在的输出目录，这是防止误删数据的保护机制。重跑任务时如果没清理旧输出，就会撞上这个错。

解决：每次重跑前删掉输出目录，或者在驱动类里用FileSystem.exists判断并自动删除/改名。我一般习惯在脚本里加一行hdfs dfs -rm -r -f $OUTPUT，跑之前先清干净。

5.3 Mapper 输出类型与 Reducer 输入类型不匹配

现象：任务运行到 Reduce 阶段抛ClassCastException，提示某个类型不能转成另一个类型。

原因：setMapOutputKeyClass/setMapOutputValueClass和 Mapper 实际输出的类型不一致，或者 Reducer 的输入类型声明和 Mapper 输出对不上。Hadoop 在运行时才做类型检查，编译期发现不了。

解决：仔细核对 Mapper 的泛型参数、context.write的实际类型、驱动类里的setMapOutput*Class三者是否一致。Reducer 的输入泛型也要和 Mapper 输出一致，差一个类型都会在 Shuffle 阶段炸出来。

5.4 脏数据导致任务中断或结果失真

现象：任务跑到一半失败，日志里是NumberFormatException；或者任务成功但结果里某些日期缺失、某些值明显异常。

原因：原始日志里有空行、表头、字段缺失、气温字段带非数字字符（比如N/A、--）等情况，Mapper 解析时没做防御，直接抛异常或者把错误值传下去。

解决：Mapper 里加空行跳过、字段数校验、try-catch包裹解析逻辑，解析失败时用计数器记录而不是抛异常。跑完任务看WEATHER.BAD_TEMP计数器，占比高就说明数据质量问题严重，得先做一轮数据清洗再跑统计。

5.5 Combiner 配错导致结果错误

现象：任务成功，但最大值、最小值、平均值和手工计算结果对不上，偏差没有规律。

原因：Combiner 的聚合逻辑和 Reducer 不一致，或者 Combiner 输出类型和 Mapper 输出类型不匹配。比如 Combiner 只保留了最大值，Reducer 却期望收到所有原始值来算最小值，结果自然错。

解决：确认 Combiner 的聚合函数满足结合律和交换律，输出类型和 Mapper 输出类型完全一致。平均值这类不满足结合律的指标，要么不用 Combiner，要么把中间结果改成sum,count复合结构。不确定的时候先把setCombinerClass注释掉，跑通再逐步加优化。

6. 二次开发：把这套模板改成自己的统计任务

这套代码的价值不在气象统计本身，而在它把 MapReduce 的输入解析、中间键值对设计、聚合逻辑、驱动配置这条链路完整走了一遍。拿到手之后，最常见的改造方向是换数据源和换统计指标。比如把气象日志换成电商订单日志，Mapper 里解析的字段从「日期、气温」改成「用户 ID、订单金额」，Reducer 里从求 max/min/avg 改成求 sum 和 count，驱动类的输入输出路径一改，就是一个按用户统计消费总额的任务。改造时唯一要重新想清楚的是 key 的选择——key 决定了数据怎么分组，分组错了聚合结果一定错。

第二个改造方向是输出格式。现在 Reducer 输出的是逗号分隔的字符串，如果下游要接 Hive 表或者导入数据库，可以改成输出Text加制表符分隔，或者自定义Writable实现序列化。改输出类型时记得同步改驱动类的setOutputKeyClass和setOutputValueClass，这两处不一致是新手最容易漏的地方。

第三个方向是加多阶段 Job。单阶段 MapReduce 只能做一次聚合，如果要做「先按天统计、再按月汇总」这种多级聚合，就得串两个 Job，前一个 Job 的输出作为后一个 Job 的输入。串联的时候注意前一个 Job 的输出目录不能和后一个的输入目录重名，中间结果目录用完及时清理，不然 HDFS 上会堆一堆临时数据。我一般会在驱动类里把中间目录命名成output-stage1、output-stage2这种带阶段编号的名字，跑完统一清理。

验证改造结果的方法和前面一样：准备小份样例数据，本地伪分布式跑一遍，手工算一遍对比，计数器看脏数据比例，输出行数对一下分组数量。这套流程走顺了，改任何统计任务都是换汤不换药。从那以后我每次改 Mapper 的解析逻辑，都强制先用几十行样例数据在本地跑一遍再上集群，省得在集群上等半天才发现字段切错了。希望帮到你。

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