简介：一套完整覆盖Hadoop气象数据处理全流程的工程代码，面向大数据课程设计、毕业设计以及需要快速上手MapReduce开发的初学者。项目基于HDFS分布式文件系统读取原始气象数据，通过MapReduce完成数据清洗、气温统计、极值分析等计算任务，并整合SSM框架将分析结果以网页形式输出，清晰展示了从存储、计算到展示的完整技术链路。代码包含任务类、业务层、持久层及前端页面，结构分层明确，方便理解作业提交、参数传递和结果封装等核心环节，同时附有数据库脚本和工程配置，导入集成开发环境即可运行调试。资源共收录562个文件，以Java源码、Jsp页面、可扩展标记语言配置、层叠样式表、脚本文件及jar依赖、class文件为主，兼顾前端交互、后端逻辑与数据库设计，压缩包整体大小约34.88兆字节，目录组织规范，支持按模块逐段学习。目前已有10577人学习下载，适合需要参考完整实现、排查运行问题或在此基础上扩展气象预测功能的开发者。
1. Hadoop分析气象数据：难点从来不在Hadoop，而在数据本身
拿到这个标题，第一反应是“气象数据上跑 MapReduce，网上 demo 一抓一大把”。但实际上，照着 WordCount 的思路去套气象观测数据，第一轮作业就会挂在数据解析上——站点文件里混着定长格式、华氏度乘以 10 存储的温度、9999 缺测标记、还有质量码，任何一个没处理干净，统计出来的月均气温都能偏差十几度。这篇笔记把一条完整链路讲清楚：气象观测文件长什么样、哪些统计适合放进 MapReduce、自定义 Writable 怎么设计、作业跑挂了怎么查。读者对象是正在拿气象或类似传感器数据练手的数据方向开发者，把这套逻辑吃透，换到其他批数据处理场景，套路完全通用。
2. 气象数据预处理与任务拆分：拿到原始 CSV 后的三件事
2.1 先看清数据再写代码：气象站观测文件到底长什么样
气象观测数据最常见的来源是按站点拆分的小文件，每个文件几十到几百 KB，记录的是某个站点逐小时或逐 3 小时的观测值。字段一般长这样：
90701,2024010112,275,-9999,10135,175,35,10,0.0 90701,2024010113,281,269,10132,180,42,12,0.0 90701,2024010114,286,272,10128,190,58,18,0.2 A3231,2024010112,310,300,10098,220,70,25,1.5我一般按这个顺序解析：站点 ID、观测时间（YYYYMMDDHH）、温度、露点、海平面气压、风向、风速、阵风、降水量。这里有两个反直觉的设计：温度是华氏度乘以 10 的整数，275 表示 27.5 华氏度；缺测值要么是 9999，要么是带符号的 -9999，不同数据平台习惯不一样。直接拿 275 当摄氏 27.5 度去统计，结果会离谱到没法看。
拿到一份新数据，别急着写 Mapper，先做三个动作：确认文件是 CSV 还是定长空格分隔；确认温度单位是摄氏度还是华氏度、有没有乘以 10；确认缺测标记是 9999 还是空字符串。这三项信息不统一，后面每一步都会踩坑。定长格式的文件往往字段中间多个空格，用split(",")解析出来的数组长度是 1，整个 Mapper 直接罢工。
2.2 哪些分析任务适合进 MapReduce：任务拆分比写代码更决定成败
气象数据分析里，不是所有问题都值得写 MapReduce。适合的任务是那种“一个键对应一组记录，做一次聚合就能出结果”的统计：按站点和月份求平均温度、按年统计极端温度、统计各站点降水量大于阈值的天数、计算日温差分布。这些任务天然分桶，shuffle 阶段把相同键的数据合并到同一个 Reducer，逻辑最顺。
不适合的任务也很明确：需要全量排序后取 Top N 的、需要跨多条记录做时间序列递推的、需要多次迭代才能收敛的，用 MapReduce 写会非常痛苦。比如“找出连续 7 天温度持续上升的站点”，这个逻辑需要访问一条记录的前后多天数据，Mapper 要缓存大量上下文，Reducer 端的处理复杂度也上去了，不如用 Hive 或者干脆导出子集到本地用 Pandas 算。
我把常见气象分析任务和落地方式整理过一个对照表：
| 分析需求 | 是否适合 MapReduce | 原因 | 更优替代 |
|---|---|---|---|
| 各站点月均温度、极值 | 非常适合 | 按站点+月份聚合一次完成 | Hive SQL |
| 某天全国高温站点 Top20 | 不太适合 | 全量排序跨 Map 阶段 | Spark RDD 的 takeOrdered |
| 连续多日升温事件识别 | 不适合 | 需要相邻记录上下文 | Spark 窗口函数 |
| 年降水总量统计 | 非常适合 | 键值聚合，天然分桶 | Hive SQL |
| 站点间温度相关性 | 边缘适合 | 需要笛卡尔积 | 预先做重分区后 join |
这个拆分的意义在于：别一上来就把整个需求塞进一个 Job。我见过不少同学把“月均温”和“年极值”混在一个 Reducer 里算，OutputValue 里塞了五六个字段，最后单位转换和格式化全糊在一起，跑出来的数据自己都不敢信。
2.3 为什么写 MapReduce 而不是 Spark：选型背后是数据规模和运维成本
现在很多团队面对气象数据第一反应是“直接用 Spark 不香吗”。我的判断标准很简单：数据量在 TB 以下、分析频率是每日定时批处理、集群里已经跑了若干 Hadoop 生态组件，那就没必要为这个任务专门引入 Spark。MapReduce 虽然慢，但逻辑直观、依赖少、作业失败重跑的成本低，一个 hadoop jar 命令就能提交，运维上没有额外负担。
反过来，如果统计维度很多、需要交互式探索、或者后续要接实时补报数据，Spark 的优势就体现出来了。还有一种常见折中：先用 MapReduce 把原始数据跑成按站点-月份聚合的中间结果，再导到 Hive 里做即席查询。气象数据属于“写多读少、统计口径稳定”的批数据，MapReduce 做清洗和粗聚合非常合适，这不丢人。
选型上还有一个容易被忽略的点：气象站点数据经常是几千个小文件，MapReduce 的 TextInputFormat 会给每个文件至少分配一个 MapTask，几千个 100KB 的文件能撑出几千个 MapTask。这一点写在第 4 章的避坑部分，真实环境里它比计算逻辑本身更能拖垮作业。
3. 完整版 MapReduce 核心代码：从自定义 Writable 到 Driver 逐类拆解
3.1 自定义 Writable 的正确写法：WeatherRecordWritable 与序列化顺序
Hadoop 的 Mapper 输出键值对经过 shuffle 时要序列化到磁盘，Text 和 IntWritable 只能表达单字段。气象记录有站点、时间、温度、质量码四个字段，最干净的做法是定义一个 Writable。核心逻辑如下：
import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class WeatherRecordWritable implements Writable { private Text stationId = new Text(); private IntWritable date = new IntWritable(); // YYYYMMDDHH private IntWritable temperature = new IntWritable(); // 原始值，华氏度 × 10 private IntWritable quality = new IntWritable(); // 质量码，0 表示可信 public WeatherRecordWritable() {} public WeatherRecordWritable(String stationId, int date, int temperature, int quality) { this.stationId.set(stationId); this.date.set(date); this.temperature.set(temperature); this.quality.set(quality); } @Override public void write(DataOutput out) throws IOException { stationId.write(out); date.write(out); temperature.write(out); quality.write(out); } @Override public void readFields(DataInput in) throws IOException { stationId.readFields(in); date.readFields(in); temperature.readFields(in); quality.readFields(in); } public Text getStationId() { return stationId; } public IntWritable getDate() { return date; } public IntWritable getTemperature() { return temperature; } public IntWritable getQuality() { return quality; } }这段代码里最重要的一行注释是 write 和 readFields 的字段顺序必须完全一致。反序列化是顺序读字节的，先写 Text 再写 Int，读的时候也得先读 Text 再读 Int，写反了温度会变成几十万。我见过团队在新加一个字段后，把 quality 插在了 temperature 前面，结果线上作业不报错，但输出数据的温度和露点全串了位，这类问题最让人头疼——MapReduce 不报错，只能靠抽样核对原始数据才发现。
3.2 解析与过滤：把脏值挡在 Mapper 的入口之外
气象数据里最脏的部分不是格式，而是语义上的“脏值”。温度 9999 表示缺测，-9999 表示漏报，质量码非 0 的数据表示没有通过质量控制，这些都必须在地图端过滤掉。解析器单独抽成一个类，方便单元测试：
public class WeatherRecordParser { public static final int MISSING = 9999; public static WeatherRecordWritable parse(String line) { String[] f = line.split(","); if (f.length < 4) { // 定长空格文件 split 出来的长度是 1，直接丢弃 return null; } try { String stationId = f[0].trim(); int date = Integer.parseInt(f[1].trim()); int temperature = Integer.parseInt(f[2].trim()); int quality = Integer.parseInt(f[3].trim()); // 质量码非 0：数据未经质量控制或已标记异常 if (quality != 0) { return null; } // 缺测标记，不同数据源可能是 9999 或 -9999 if (temperature == MISSING || temperature == -MISSING) { return null; } // 温度合理性检查：华氏度 × 10 后，全球极端气温约在 -1500 到 1500 之间 if (temperature < -1500 || temperature > 1500) { return null; } return new WeatherRecordWritable(stationId, date, temperature, quality); } catch (NumberFormatException e) { return null; } } }这里的过滤顺序有讲究：先看质量码，再看缺测标记，最后做范围检查。质量码过滤放在最前面，因为质量码异常的数据即使温度值看起来正常也不该用。范围检查的边界值我一般保守放大一些，南极内陆最低温约零下 89 摄氏度，折合成华氏度乘 10 约 -1282，如果边界只到 -1000 就会把正常低温数据误杀。
3.3 Mapper、Combiner 与 Reducer 的代码实现与单位转换设计
统计目标是“每个站点每个日历月的平均温度、最高温度和最低温度”。Mapper 阶段按站点加月份聚合，把日期整数除以 100 得到 YYYYMM 作为键的一部分。Combiner 单独写一个类，它的冒烟测试题就是“不要复用 Reducer”：
import org.apache.hadoop.io.*; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; // 用于 Mapper 输出和 Combiner 输出的统计中间量 class TemperatureStatWritable implements Writable { private LongWritable sum = new LongWritable(); private IntWritable count = new IntWritable(); private IntWritable max = new IntWritable(Integer.MIN_VALUE); private IntWritable min = new IntWritable(Integer.MAX_VALUE); public TemperatureStatWritable() {} public TemperatureStatWritable(int temperature) { sum.set(temperature); count.set(1); max.set(temperature); min.set(temperature); } // readFields 和 write 方法省略，字段顺序为 sum、count、max、min } public static class TempMapper extends Mapper<Object, Text, Text, TemperatureStatWritable> { private Text outKey = new Text(); @Override protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { WeatherRecordWritable record = WeatherRecordParser.parse(value.toString()); if (record == null) { return; } // date 是 YYYYMMDDHH，除以 100 得到 YYYYMM，作为月份聚合键 int month = record.getDate().get() / 100; outKey.set(record.getStationId().toString() + "-" + month); int t = record.getTemperature().get(); context.write(outKey, new TemperatureStatWritable(t)); } }Mapper 的输出键是“站点-月份”字符串，值为单条记录的统计封装。Combiner 只在 Map 端做局部归并，显著减少 shuffle 到 Reducer 的数据量，这在气象数据里非常有效，因为同一个站点同一个小时可能有多次观测，Combiner 能把几千条记录压成一条：
public static class TempCombiner extends Reducer<Text, TemperatureStatWritable, Text, TemperatureStatWritable> { @Override protected void reduce(Text key, Iterable<TemperatureStatWritable> values, Context context) throws IOException, InterruptedException { long sum = 0; int count = 0, max = Integer.MIN_VALUE, min = Integer.MAX_VALUE; for (TemperatureStatWritable v : values) { sum += v.sum.get(); count += v.count.get(); max = Math.max(max, v.max.get()); min = Math.min(min, v.min.get()); } TemperatureStatWritable result = new TemperatureStatWritable(); result.sum.set(sum); result.count.set(count); result.max.set(max); result.min.set(min); context.write(key, result); } }Combiner 和 Reducer 的输入输出类型完全一样，但这里刻意不用同一个类当 Combiner，原因在下一条 Reducer 代码里体现：
public static class TempReducer extends Reducer<Text, TemperatureStatWritable, Text, Text> { private Text outValue = new Text(); @Override protected void reduce(Text key, Iterable<TemperatureStatWritable> values, Context context) throws IOException, InterruptedException { long sum = 0; int count = 0, max = Integer.MIN_VALUE, min = Integer.MAX_VALUE; for (TemperatureStatWritable v : values) { sum += v.sum.get(); count += v.count.get(); max = Math.max(max, v.max.get()); min = Math.min(min, v.min.get()); } // 温度转换：原始值是华氏度 × 10，先还原华氏度，再转摄氏度 // (fahrenheit * 10 - 320) / 18 等价于 (fahrenheit - 32) * 5 / 9 double avgC = (sum / (double) count - 320.0) / 18.0; double maxC = (max - 320.0) / 18.0; double minC = (min - 320.0) / 18.0; String statistic = String.format("avgC=%.1f maxC=%.1f minC=%.1f count=%d", avgC, maxC, minC, count); outValue.set(statistic); context.write(key, outValue); } }如果 Combiner 复用 Reducer，Map 端就会提前把华氏度转成摄氏度，Reducer 拿到的是已转换的数据，再减 320 除以 18，结果歪到不可收拾。单位转换只能发生在最终输出阶段，中间态的 sum、count、max、min 必须保持原始华氏度整数值。这条原则写进代码注释里，能帮后来的维护者省掉大量排错时间。
3.4 Driver 与作业提交命令：参数配置的取舍
Driver 负责组装整个作业，并设置关键的调优参数：
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class WeatherAnalyzer { public static void main(String[] args) throws Exception { if (args.length != 2) { System.err.println("Usage: WeatherAnalyzer <input> <output>"); System.exit(-1); } Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "weather-monthly-stats"); job.setJarByClass(WeatherAnalyzer.class); job.setMapperClass(TempMapper.class); job.setCombinerClass(TempCombiner.class); job.setReducerClass(TempReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(TemperatureStatWritable.class); // 输入格式不用改，TextInputFormat 一行一个 key-value job.setNumReduceTasks(8); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }setNumReduceTasks(8)不是拍脑袋定的。气象站点数据天然有热点，大城市周边站点密度远高于无人区，Reduce 任务数太少会导致某个站点所在的 Reduce 长时间卡在 99%，Reduce 任务数太多又增加 shuffle 的小文件数量。默认值是 1，气象场景下基本都会调大。要注意setOutputValueClass设的是 Mapper 的输出类型，不是 Reducer 最终输出类型，Reducer 最终输出 Text 写在context.write里，这个类型不用在setOutputValueClass里声明。
编译打包后提交命令：
hadoop jar weather-analysis.jar com.example.WeatherAnalyzer \ /data/weather/raw /data/weather/monthly-stats输出目录必须是不存在的路径，否则作业直接报错。这是 Hadoop 的防误覆盖设计，不是 bug。重跑作业的习惯做法是先删旧输出目录再提交，或者把输出目录带上时间戳，例如/data/weather/monthly-stats/20240101，这样每次跑批的结果都留档，后续对比口径方便得多。
4. 气象数据 Hadoop 作业的翻车现场：5 个高频坑与排查路径
4.1 温度字段被解析成乱码：分隔符和定长格式混用
现象：Map 阶段大量 Task 失败，日志里全是NumberFormatException，但看原始文件明明都是数字。
原因：数据平台导出的文件混用了两种格式，一部分是标准 CSV，一部分是空格定长。定长文件里每行只有一两个连续空格分隔，split(",")把整行当成一个字段，parse 时所有字段全部落到第一个元素，后续索引越界或直接抛异常。这种文件在 HDFS 上按站点目录存放，不同站点的采集设备可能来自不同批次，格式不统一是常态。
解决：解析器里先判断逗号数量，如果split(",")结果长度小于预期，就退化用split("\\s+")按空白切分。还有一个更稳妥的做法是预处理阶段用脚本把所有文件统一转成标准 CSV 再上传，把格式问题挡在 Hadoop 作业之外。我在生产环境里选的是后者，因为 Mapper 里做格式探测会影响每条记录的解析速度，而预处理只需跑一次。
String[] f; if (line.contains(",")) { f = line.split(","); } else { f = line.trim().split("\\s+"); }4.2 统计结果里突然出现 9999 度：缺测值没过滤干净
现象：月均温统计结果正常，但最高温一栏出现 9999 或 -9999，数据直接穿帮。
原因：解析器只过滤了质量码非 0 的数据，但有些站点的缺测值质量码是 0，表示“这个位置本应有一个值，但采集器没读到”，语义属于合法占位符，9999 本身就在质量合格的记录里出现了。
解决：过滤条件必须同时看质量码和温度值，缺测标记 9999 与 -9999 单独拦截。检查代码时可以用一个简单技巧：统计结果里如果出现过 9000 以上的值，第一反应不是检查计算逻辑，而是去原始数据里搜 9999 的分布。这条经验是我从一次线上事故里学来的，那个作业跑了两个月，后来核对历史数据才发现有一整个月的最高温被 9999 污染了。
4.3 数据倾斜：热点站点的 Reduce 卡在 99%
现象：作业整体进度停在 Reduce 99% 超过半个小时，部分 Reduce 任务完成，个别任务迟迟不结束。
原因：气象站点数据分布极不均匀，大城市站点的观测频率和记录条数可能是偏远站点的几百倍。按站点-月份作为键，热点站点所在的 Reduce 要处理的数据量是其他 Reduce 的百倍，木桶效应直接卡死整个作业。
解决：加 Combiner 是第一层缓解，把 Map 端的重复记录先聚合。还不够的话，对热点键做加盐处理，比如把站点 ID 后面拼一个随机数后缀，让热点数据分散到多个 Reduce，最后再跑一个辅助作业把盐去掉做二次聚合。加盐的代价是多一个作业，但换来的是作业稳定不卡死。自定义 Partitioner 也可以按站点哈希分散，但效果不如加盐彻底。
4.4 Combiner 复用 Reducer 后单位转换错乱
现象：作业跑完，avgC 和 maxC 的值对不上，逐条核对原始数据发现温度被转换了两次。
原因：Reducer 里做了华氏度转摄氏度的换算，如果 Combiner 直接复用同一个类，Map 端的局部聚合就先把单位转了，Reducer 拿到已转换的数据又转一次，两轮运算后数值完全扭曲。
解决：Combiner 和 Reducer 必须分成两个类，Combiner 只做纯整数聚合，单位转换只允许在 Reducer 最终输出时发生。代码评审时把这个原则写进注释，比靠后来的人自觉靠谱得多。同类问题还会出现在增加新统计字段时，如果新字段想在 Combiner 里做预处理，要重新评估它对最终结果的影响。
4.5 几千个小文件拖垮 NameNode
现象：作业提交后 MapTask 数量爆炸，调度时间远超计算时间，集群 CPU 利用率很低，NameNode 内存告警。
原因：气象观测数据按站点拆成几千个不到 200KB 的小文件。TextInputFormat 默认每个文件至少一个 InputSplit，每个 split 对应一个 MapTask，几千个小文件直接生成几千个 MapTask，大部分时间耗在 YARN 调度和容器启动上。
解决：上传前先在本地用hdfs dfs -getmerge把同批次文件合并成几个大文件再传回 HDFS，或者写一个预处理作业把这些小文件转成 SequenceFile，让一个 MapTask 处理一个大文件。合并粒度控制在每个文件 64MB 到 128MB 之间，能压到几百个 MapTask 以内，作业整体耗时能缩短一半以上，这个优化是我在所有气象数据作业里的第一优先级。
5. 从单 Job 到验证闭环：结果核对、串联分析与下一步演进
5.1 串联两个 Job：月统计结果如何支撑站点对比分析
单 Job 算完月均温和极值之后，气象分析一般还有第二层需求：对比不同站点同一月份的温度差异，或者找出年温差最大的站点。第二层需求的输入恰好是第一层 Job 的输出，用 JobControl 串联两个作业比写两个独立命令更可靠：
import org.apache.hadoop.util.ToolRunner; import org.apache.hadoop.util.Tool; public class WeatherPipeline implements Tool { public int run(String[] args) throws Exception { Job job1 = new Job(getConf(), "weather-monthly-stats"); // job1 配置省略，与 3.4 节一致 Job job2 = new Job(getConf(), "weather-station-compare"); // job2 的输入是 job1 的输出目录 JobControl control = new JobControl("weather-pipeline"); ControlledJob cJob1 = new ControlledJob(conf); ControlledJob cJob2 = new ControlledJob(conf); cJob2.addDependingJob(cJob1); control.addJob(cJob1); control.addJob(cJob2); Thread t = new Thread(control); t.start(); while (!control.allFinished()) { Thread.sleep(1000); } return control.getFailedJobList().isEmpty() ? 0 : 1; } }这里要强调一个工程习惯：第二个 Job 的 Reduce 输出必须是“一行一站、一列一指标”的规整格式，而不是第一个 Job 输出的那种人类可读文本。规整格式意味着后续不管接 Hive 还是接 Pandas，都能直接读，不用再做一次解析。
5.2 验证输出的笨办法与升级方向
MapReduce 作业跑完，不能只看 exit code 是 0 就收工。我一般抽样三个站点，从原始数据里手算该站点某个月的温度，对比作业输出值是否一致。抽样时要选一个数据量大的热点站点、一个普通站点、一个靠近两极的低温站点，覆盖不同数据分布。数据量大的站点能验证 Combiner 逻辑，低温站点能验证单位转换公式在负温度下是否仍然正确。
作业调优的空间还有两处：一是把输入换成 Parquet 列式存储，温度列单独读取，I/O 量能降一个数量级；二是当集群资源紧张时，用 Fair Scheduler 给天气作业单独划一个队列，避免夜间跑批和其他任务互相抢资源。这几个方向的改动都不大，但收益很直观。
最后说一个习惯：每次改完解析器或 Combiner，先用一个 1MB 的样本数据在本地跑通，再上集群全量执行。这个习惯帮我挡住过至少三次单位转换错误和两次字段顺序错位的回归。所谓“完整版代码”，不只是一堆类能编译通过，而是每次改动后都能用同一套验证方法兜底，确保输出的每个温度都经得起核对。希望这篇笔记能帮你把气象数据的 MapReduce 链路一次跑顺。
本文还有配套的精品资源，点击获取
