企业官网建设流程全解析

简介：面向正在学习Hadoop大数据技术、准备气象数据课程设计或毕业设计的开发者，这套完整代码提供了从数据预处理到结果展示的可运行工程，省去从零搭建的繁琐过程。压缩包共562个文件，大小34.88MB，包含Java源码、MapReduce处理类、SSM框架页面资源（JS/CSS/JSP/XML）、依赖JAR包、SQL脚本及配置文件等，目录按功能模块划分，便于快速定位。已有10580人浏览学习，在同类资源中具有较高参考热度，适合作为实战入门及教学示例。代码核心覆盖HDFS存储、MapReduce并行计算、气象指标统计与基于SSM的可视化展示等环节，包含温度极值、平均气温等常见统计的Mapper/Reducer实现，并可与前端页面联调。资源附带依赖库、静态资源、部署配置文件，导入IDE后即可运行调试，帮助读者理顺分布式计算与Web展示的完整链路，有效提升工程实践能力。

1. Hadoop分析气象数据：完整版代码到底解决什么问题

如果你手里有一批气象观测站的多年温度、降水量、风速数据，想统计“某年某站最高气温”“连续高温天数”“历史极值排名”，最先想到的可能是写个 Python 脚本循环读文件。但当你把文件从几百 MB 换成几十 GB、网格数据变成上千个站点分文件时，单机脚本就开始卡顿、超时、内存报错。Hadoop分析气象数据完整版代码，就是一套能直接在 HDFS 上跑、可扩展、可调度、能接手别人数据的 MapReduce 方案。它不解决“单机读一个 CSV”的问题，而是解决“数据量大到没法放进内存、还得按年份和站号聚合”的问题。适合刚学完 Hadoop 基础、想摆脱 wordcount 模板，或者正在做气象数据离线分析的开发者。接下来我会从输入数据建模开始，给你一份能落地、能改、能排错的完整实现。

2. 气象数据长什么样？分析任务该怎么拆成 MapReduce

2.1 气象数据的三种常见存储格式与字段陷阱

我接触到的气象数据源，常见有三种格式。第一种是固定列宽文本，比如某些自动站导出的文件，每行固定 80 个字符，第 10 到 15 位是气温，第 16 到 20 位是降水量。第二种是逗号分隔的 CSV，通常带表头，字段包括站号、观测时间、气温、相对湿度、气压、风速、风向、降水量。第三种是 JSON 或 Avro，多用于设备直接上报，字段名不固定。

完整的 MapReduce 代码首先要处理的是“脏数据”，不是“大文件”。实际数据里最常见的坑是：CSV 中时间字段写成2024-1-5，而不是2024-01-05；气温字段被标成9999.9作为缺测值；降水量按“0.1mm”整数存储，需要除以 10。这些如果在 Mapper 里不处理，后期所有统计结果都对不上。

所以我建议第一步不是写 Hadoop 代码，而是写一个字段校验清单：站号、日期、时刻、气温、降水、湿度、风速、气压，每列的数据类型和值域定义清楚。气象分析里最典型的任务是“按年份分组、按站点聚合”，比如求某站点的年最高气温、年累计降雨量。这就是典型的“分组加聚合”，用 MapReduce 做非常合适，因为不需要跨节点做全局排序，中间结果只有年份和站点两个维度。

2.2 把“求某年气温极值”映射成 MapReduce 的键值对

MapReduce 的编程模型本质是“换一种方式思考数据流”。对气象数据来说，你要先定义“什么是 key，什么是 value”。拿“每个站点每年最高气温”这个需求举例，我会这样设计：

• Key：站号 + 年份，合并成一个文本，形如01001-2024
• Value：该条记录的气温值，使用浮点数DoubleWritable

Mapper 的任务是：逐行读入原始文本，解析出站号、日期、气温，然后context.write(Text, DoubleWritable)。Reducer 的任务是：对同一个01001-2024下的所有气温值取最大值，写回文件。

这个设计的核心逻辑是，Map 阶段把所有数据按照维度分桶，Reduce 阶段只在桶内做计算。由于同一个 key 的所有 value 必然落到同一个 reducer，所以不需要全局归并。这样对于“年极值”这种场景，MapReduce 不需要排序中间结果，效率会高很多；但如果要求“全历史前 10 名”，就不能用这种方式，需要用到TreeMap或自定义分区，我会在最后一章讲。

2.3 选型理由：为什么 HDFS + MapReduce 比单机脚本靠谱

很多人有一个误解：MapReduce 比 Python 快。真实情况是，在 10 GB 以内的数据上，单机脚本通常更快，省掉了启动 JVM、任务调度、数据分发这些开销。MapReduce 的价值是“稳定跑大数据量”和“并行扩容”。

气象数据的特点是多文件、多站点、多时间片。自动站数据每小时生成一个文件，一年的数据量可能是数万个文件。把这些文件全部放到 HDFS 上，Hadoop 会按块分布到集群的多个节点，Map 任务会在数据所在节点上启动，尽量做到本地计算。这是单机脚本做不到的。

另外，MapReduce 对单条坏记录的容忍度很高。某个文件里有一行是乱码，只会导致一个 map task 失败并重试，不会让整个任务崩溃。你用 Python 写循环，一个UnicodeDecodeError就能把批处理拍停。这就是我会向别人推荐“气象数据离线分析优先考虑 Hadoop 生态”的原因。如果数据量不够大，用单机脚本没问题；一旦数据量大到要管文件分布，Hadoop 方案的整个架构收益就体现出来了。

3. 手把手写出完整版代码：Mapper、Reducer、Driver 一个不少

3.1 核心 Java 代码：气温极值统计的三段式

先给一份最简但可跑的 Java 代码，目标是从气象 CSV 中统计每个站点每年的最高气温。输入假设每条记录格式为：

station_id,date,maxtemp,mintemp,precipitation 01001,2024-05-01,28.5,17.2,0.0

Mapper 实现：

import java.io.IOException; import org.apache.hadoop.io.DoubleWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class WeatherMapper extends Mapper<LongWritable, Text, Text, DoubleWritable> { private Text outKey = new Text(); private DoubleWritable outValue = new DoubleWritable(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); // 跳过头行和空行 if (line.length() == 0 || line.startsWith("station_id")) { return; } String[] fields = line.split(","); if (fields.length < 4) { // 脏数据行，用计数器记录，方便观察 context.getCounter("Weather", "MalformedLines").increment(1); return; } try { String stationId = fields[0].trim(); String date = fields[1].trim(); double maxTemp = Double.parseDouble(fields[2].trim()); // 过滤缺测值 9999.9 if (Math.abs(maxTemp - 9999.9) < 0.01) { return; } // 从日期中提取年份，例如 "2024-05-01" -> "2024" String year = date.substring(0, 4); outKey.set(stationId + "-" + year); outValue.set(maxTemp); context.write(outKey, outValue); } catch (NumberFormatException e) { context.getCounter("Weather", "NumberFormatErrors").increment(1); } } }

这段 Mapper 做了三件容易遗漏的事情：跳表头、验字段数量、过滤缺测值。计数器是关键技巧，你跑完任务后在控制台能看到Weather计数器的数值，用这个判断脏数据比例。

Reducer 实现：

import java.io.IOException; import org.apache.hadoop.io.DoubleWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class WeatherReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> { private DoubleWritable result = new DoubleWritable(); @Override protected void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException { double maxValue = Double.MIN_VALUE; for (DoubleWritable val : values) { if (val.get() > maxValue) { maxValue = val.get(); } } result.set(maxValue); context.write(key, result); } }

Reducer 逻辑很直接：遍历当前 key 的所有气温值，保留最大。如果要从“最高气温”改成“最低气温”，把maxValue初始化改成Double.MAX_VALUE，判断条件改成小于即可。

Driver 实现：

import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.DoubleWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class WeatherDriver { public static void main(String[] args) throws Exception { if (args.length != 2) { System.err.println("Usage: WeatherDriver <input-path> <output-path>"); System.exit(-1); } Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "Weather Max Temp by Station-Year"); job.setJarByClass(WeatherDriver.class); job.setMapperClass(WeatherMapper.class); job.setReducerClass(WeatherReducer.class); job.setCombinerClass(WeatherReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(DoubleWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }

这里最难理解的是job.setCombinerClass(WeatherReducer.class)。Combiner 的作用是在 map 输出后、网络传输前，先在本地做一次局部聚合。对求极值这种操作，Combiner 和 Reducer 逻辑完全一致，所以可以直接复用 Reducer 类。如果你的分析逻辑换成了求平均值，就不能直接复用，需要单独写 Combiner，原因我在下一章讲。

3.2 关键参数说明：JVM 内存、Reduce 数、输入格式

代码能跑只是第一步，参数调不对，大任务照样失败。

• mapreduce.map.memory.mb和mapreduce.reduce.memory.mb：控制 map/reduce 容器的内存上限。气象数据单条记录本身很小，但 Mapper 如果一次读入过大的缓冲，容易出现 OOM。我一般把 map 内存设为 2048，reduce 内存设为 4096，单位是 MB，具体看集群规格。
• mapreduce.job.reduces：控制 Reduce 数量。很多人以为越多越好，实际不是。如果只有 100 个键，设 1000 个 reducer 会产生大量小文件。建议按数据量估算，一般 1 GB 输入数据配 4 到 8 个 reducer 比较稳。
• mapreduce.input.fileinputformat.split.maxsize：控制输入分片大小。如果你的气象站文件全是小文件，这参数就已经救不了你了，要靠CombineFileInputFormat，我放在避坑章节讲。

3.3 本地跑通的最小命令与集群提交命令

本地模式最快，但要注意本地模式不会启动 YARN。开发阶段我会把输入数据放到本地目录，然后执行：

hadoop jar weather-analysis-1.0.jar WeatherDriver /home/user/weather_data/2024_monthly.csv /home/user/weather_output

如果直接跑在集群上，输入输出路径要改成 HDFS 路径：

hdfs dfs -mkdir -p /logs/weather/2024_input hdfs dfs -put /home/user/weather_data/2024_monthly.csv /logs/weather/2024_input/ hadoop jar weather-analysis-1.0.jar WeatherDriver hdfs://namenode:8020/logs/weather/2024_input hdfs://namenode:8020/logs/weather/2024_output

注意/logs/weather/2024_output在运行前不能存在。这是 MapReduce 的一个硬性约定，避免覆盖上一次结果。如果第二次想重跑，必须执行hdfs dfs -rm -r /logs/weather/2024_output。这个坑我踩过很多次，后来都在脚本里加了一行先删除旧输出目录。

4. 从简陋到能用的版本演进：多指标统计、Hive 与调度

4.1 扩展一：Combiner 做同一个 station 的多指标聚合

气象分析不会只统计最高气温，你往往要同时算平均气温、累计降水量、最大风速。这时候如果你在 Mapper 里把每个指标都context.write一遍，一定会出现数据倾斜：有些站点数据量大，对应的 key 流量特别高。我的做法是，把多个指标封装到一个Writable对象里，在 Map 阶段输出一次，Reducer 里再拆开。

但这样 Combiner 就不能直接复用 Reducer 了。因为 Combiner 要合并的是“同一批次内多个指标的部分结果”，例如你需要平均值时，Combiner 里就得维护“计数”和“总和”，然后输出一个自定义的StationSummaryWritable。这时 Reducer 在收到 Combiner 结果后，再合并所有 map 端聚合结果。写这类代码时，我建议给StationSummaryWritable实现Writable接口，必须重写write和readFields，字段顺序要保持一致，否则集群上反序列化会报错。这是很多初学者最容易翻车的地方。

4.2 扩展二：MultipleOutputs 按年份分区写出结果

默认情况下，Reducer 的输出会写到一个part-r-00000文件里。如果你想按年份分别输出，比如2023_max_temperature.txt、2024_max_temperature.txt，不需要写多个 reducer，使用MultipleOutputs就行。

Driver 里注册命名输出块：

import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; Path outPath = new Path(args[1]); MultipleOutputs.addNamedOutput(job, "year", TextOutputFormat.class, Text.class, DoubleWritable.class); FileOutputFormat.setOutputPath(job, outPath);

Reducer 里不再context.write，改调用multiOut.write(key, value, "year_" + year)。这样输出文件会按year_2024-r-00000的形式命名。这个功能在后续把结果导入 Hive 或分区表时特别有用，能直接从输出路径做分区挂载。

注意一个坑：MultipleOutputs必须在cleanup()里关闭，否则数据不会刷到磁盘。这个和context.write完全不同，忘了 close 会出现“任务成功但输出空文件”的诡异现象。

4.3 扩展三：用 Hive 写 SQL 替代重复造轮子

如果你已经用了 Hive 作为数据仓库层，那么 Hadoop 气象分析代码完全可以简化成 HiveQL。我经常见到同事纠结要不要写 MapReduce，其实 Hive 会把 SQL 翻译成 MapReduce，所以没必要重复造轮子。

创建气象表：

CREATE EXTERNAL TABLE weather_ods ( station_id STRING, record_date STRING, max_temp DOUBLE, min_temp DOUBLE, precipitation DOUBLE ) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LOCATION '/warehouse/weather_ods/';

然后直接查年最高气温：

SELECT station_id, SUBSTRING(record_date, 1, 4) AS yr, MAX(max_temp) FROM weather_ods GROUP BY station_id, SUBSTRING(record_date, 1, 4);

商业环境下，我会优先用 Hive 做这种简单聚合，因为可读性好、别人好维护。只有当 HiveQL 写不出复杂的自定义逻辑，或者需要精确控制 Map 阶段的数据清洗流程，才会回到底层 MapReduce。你要把两者看成互补，不是二选一。

4.4 把分析任务挂进调度：从手动提交到可视化依赖

气象数据的分析通常是周期性的，每天新增一批数据跑一次日统计，月底跑月度汇总。如果你每天手动敲hadoop jar命令，三个月后一定会漏跑。我建议至少写一个 shell 脚本包住提交流程，再用调度器按时触发。

一个标准的日批处理脚本长这样：

#!/bin/bash source /etc/profile TODAY=$(date +%Y%m%d) INPUT_PATH="/logs/weather/raw/$TODAY" OUTPUT_PATH="/logs/weather/summary/$TODAY" hdfs dfs -test -e "$INPUT_PATH" if [ $? -ne 0 ]; then echo "input not found: $INPUT_PATH" exit 1 fi hdfs dfs -rm -r "$OUTPUT_PATH" 2>/dev/null hadoop jar /data/apps/weather-analysis.jar \ WeatherDriver "$INPUT_PATH" "$OUTPUT_PATH" exit $?

脚本里先判断输入是否存在，再清理历史输出目录，最后用exit $?把运行状态传递给调度器。如果你的集群装了 Oozie 或 Airflow，就把整个脚本当一个 action 节点，流程里加数据完成依赖。这样“完整版代码”才算是工程闭环，而不是一个能跑一次的程序。

5. 避坑清单：气象数据跑 Hadoop 的六个典型翻车现场

5.1 小文件一多，Map 数量爆炸

现象：输入数据有 3 万个气象站小文件，每个文件只有几十 KB，任务启动以后出现了 3 万个 map task，整个集群卡在“RUNNING”状态，资源开销巨大，跑了半小时还没结束。

原因：FileInputFormat默认按文件拆分，一个文件至少一个 map task。气象自动站恰好每小时生成一个文件，一年累计下来海量小文件，分片数远远超过集群可用资源。

解决：不要直接用TextInputFormat，改用CombineFileInputFormat。它能把多个小文件打包成一个逻辑分片，然后再交给一个 map task 处理。配置示例：

import org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat; job.setInputFormatClass(CombineFileInputFormat.class); CombineFileInputFormat.setMaxInputSplitSize(job, 67108864);

如果数据源还在产生新文件，更彻底的办法是提前把小文件合并成大文件，比如用hdfs dfs -getmerge或者 Hive 里的INSERT OVERWRITE ... SELECT。我会优先建议合并，因为不只是 Map 数量减少，后续所有分析任务都受益。

5.2 读进来全是 NULL：分隔符和 CSV 引号没处理

现象：Mapper 里line.split(",")后字段数总是比预期少，某些行读取出来是[null, null]，或者气温字段解析成默认值 0.0。

原因：气象 CSV 里经常出现带引号的字段，比如"温度"或"2024-05-01 00:00"。如果日期字段被引号包裹，逗号分隔后引号也参与解析，导致 fields 数组里多个引号符号。另外有些文件是分号分隔，或者制表符分隔，你没有提前确认。

解决：在 Mapper 里统一清洗一行，而不是假设字段格式。我的习惯是先把整行做一次正则校验，然后去引号、去空格。对于超大 CSV，尽量避免引入复杂的解析库，简单字段用split(",")后手动trim()和replace("\"", "")足够。所有清洗逻辑要写在一处，不要散落在不同方法里。

5.3 Reduce 端内存溢出：DistributedCache 与堆参数

现象：任务运行到reduce阶段，日志显示GC overhead limit exceeded或直接OutOfMemoryError: Java heap space。

原因：气象分析中，如果 Mapper 输出大量键值对，Reducer 需要把属于同一 key 的所有 value 拉到内存里迭代。某几个站点数据量特别大，导致单个 reducer 内存被击穿。另一个常见原因是你在 Mapper 里读了DistributedCache的大文件，但配置的缓存文件没有设上限，导致容器启动时把所有缓存文件加载到内存。

解决：优先用mapreduce.reduce.reduce.input.buffer.percent调整 reducer 端 shuffle 内存在堆中的占比，默认是 0.0，表示 shuffle 内存由框架管理。我通常把它设成 0.2 到 0.4 之间。同时设置：

-Dmapreduce.reduce.memory.mb=4096 -Dmapreduce.reduce.java.opts=-Xmx3g

注意mapreduce.reduce.java.opts是 JVM 堆设置，mapreduce.reduce.memory.mb是容器内存。两者要留有差值，给非堆留下空间。气象数据笛卡尔积场景很少，如果出现某个 key 过大，更合理的方案是在 Mapper 侧提前加 Combiner，减少传输量。

5.4 日期字符串按字典序排序，导致极值算错

现象：统计历史最高气温，结果却出现了9999这种明显缺测值，或者某年 1 月的数据全部没进结果。

原因：有些气象数据把月份写成字符串，比如日期字段是2024-01-05，你用substring(0, 4)提取年份没问题；但如果源文件里年份字段是2024-Jan-05这种格式，你用字符串截取就会截出2024-J，后续全都算错。还有一个隐蔽场景：日期字段被写成2024-5-5，substring(0, 4)取到了2024-，年份变成空字符串。

解决：不要用“位置截取”，要用日期工具类解析。在 Mapper 里我会直接用SimpleDateFormat或 Java 8 的LocalDate解析整个日期字段，然后取getYear()。如果解析失败，这条数直接算脏数据，通过计数器计数。特别提醒，气象数据中缺测日期经常被填成9999-99-99，解析日期前要先用正则过滤掉这类非法值。

5.5 本地正常集群失败：时间同步、文件权限和目录残留

现象：本地hadoop jar跑得好好的，一上集群就报Permission denied或FileAlreadyExistsException，而且反复重试也稳定出错。

原因：本地模式下很多权限检查被绕过，集群模式下 HDFS 会严格校验用户身份。另一个常见原因是上一次任务失败后，输出目录没有删除，新任务启动时发现同名目录，直接拒绝运行。

解决：在 Driver 里预检查输出路径，如果存在则主动删除，这是一种相对安全的做法。另外在提交命令里加上执行用户的--principal或确保当前用户对 HDFS 目录有写权限。如果集群启用了 Kerberos，还要先kinit拿到票据。我的习惯是，把权限检查放到脚本最前面，宁可失败在提交前，也不要失败在任务跑了一半。

6. 让代码更稳的进阶套路：计数器、单元测试与 Top-N 输出

一个工程化项目，完整版代码不只是功能能跑，还要可观测、可回归、能处理更复杂的业务需求。我的最后一个压箱底技巧是：在 Mapper 里加数据质量计数器，用getCounter把坏行、缺测值、日期解析失败数全部统计出来。不要觉得这些是小事，真实气象数据里脏数据比例经常超过 5%，没有计数器你根本不知道自己的统计结果到底丢了多少记录。

进阶一点的聚合场景，比如你要输出“历年最高气温 Top 3”，MapReduce 默认的 shuffle 排序只能按 key 排序，不能直接输出桶内 Top-N。常见做法是 Mapper 内部用TreeMap维护每个站号 Top-N 的气温记录，然后再输出。这样能把单个 map 任务内的数据量压缩到 3 条，极大减少传输。伪代码如下：

TreeMap<Double, String> topMap = new TreeMap<>(); // 在 map 中维护 top N topMap.put(temp, stationId + "-" + date); if (topMap.size() > 3) { topMap.remove(topMap.firstKey()); } // cleanup 中输出 topMap

这种技巧的代价是，你失去了“全局 Top-N”的精确性，因为不同 map 任务之间没有互相通信。如果想要全数据准确 Top-N，就需要用自定义分区、全排序或辅助排序，复杂度再上一个台阶。实际业务里，我通常按“每年先算站点极值，再做全局 Top-N”的两阶段任务，用一次 MapReduce 解决不下所有问题。

回到开头那句话，所谓完整版代码，不是把一份代码从 wordcount 改个名字，而是知道每一步为什么这样写：Maper 里的清洗逻辑、Combiner 的复用条件、输出的分区策略，以及最终如何调度收口。我最初跑气象数据时，在缺失值处理上吃过亏，后来每到新项目都会先写数据质量清单，再动业务逻辑，少走很多弯路。希望帮到你。

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