简介这份资源是面向高校《分布式》课程学习者与Hadoop入门开发者的完整作业级项目源码围绕气象数据分析场景解决从数据清洗统计到结果可视化展示的全流程实现问题。包内以Hadoop MapReduce程序为核心完成气象数据的分布式处理与指标计算同时配套SSM框架代码将分析结果通过Web页面呈现形成离线计算加前端展示的闭环方案。压缩包为rar格式整体约34.89MB文件类型以Java源码、配置文件及项目工程文件为主分别对应MapReduce任务逻辑、Spring与MyBatis整合配置以及页面展示模块。目前已有3294人学习下载适合作为课程作业参考或分布式入门练手项目。读者可据此理解气象数据在HDFS上的存储与切分方式、Mapper与Reducer的编写要点、任务提交与运行流程以及如何用SSM把统计结果接入数据库并展示同时借鉴项目目录组织与配置排错思路快速搭建可运行的分析环境。1. 气象数据遇上 Hadoop一套能跑通的离线分析系统长什么样气象数据有个很别扭的特点单站点每天产生的观测记录不大但全国几千个站点、连续几十年的分钟级数据堆在一起就是几百 GB 到 TB 级别。更麻烦的是气象业务里最常见的查询不是「查某一条」而是「按站点、按时间段做聚合统计」——月平均气温、年降水量极值、风向频率分布。这种场景用单机 MySQL 硬扛数据量一上来查询就慢得让人想砸键盘。这套「Hadoop 分析气象数据完整版源代码」要解决的正是这个问题用 HDFS 存原始气象记录用 MapReduce 做离线批量统计再用 SSMSpring SpringMVC MyBatis搭一个 Web 层把分析结果展示出来。它适合两类人一是做 Hadoop 课程设计、需要一套完整可运行项目的学生二是想理解「离线计算 Web 展示」这条链路怎么打通的初中级工程师。整条链路的核心不是某个单点技术而是数据从原始文本到页面图表的流转过程中间任何一环配置错了结果就是跑不出来或者跑出来是错的。2. 数据怎么进 HDFS气象原始记录的清洗与上传2.1 气象数据的典型格式与字段设计真实气象观测数据常见两种格式一种是 CSV每行一条记录字段用逗号分隔另一种是固定宽度文本每个字段占固定列数。不管哪种进 HDFS 之前最好先统一成一种结构清晰的格式否则 MapReduce 里解析逻辑会写得非常痛苦。我一般会把原始数据整理成这样的字段顺序字段名含义示例类型station_id站点编号54511Stringdate观测日期2023-01-15Stringtime观测时刻14:00Stringtemperature气温℃3.5Doublehumidity相对湿度%42Intpressure气压hPa1013.2Doublewind_speed风速m/s2.8Doublewind_direction风向度180Intprecipitation降水量mm0.0Double这个字段设计不是随便定的。station_id 和 date 是后续 MapReduce 分组的关键temperature、humidity、precipitation 是统计对象wind_direction 用角度存是为了方便做风向频率的区间统计。如果原始数据里风向是「东南风」这种文字描述进 HDFS 前一定要转成角度值否则 MapReduce 里做区间划分会多出一堆字符串匹配的代码。清洗环节用 Python 写个脚本就够了不需要上什么重型工具import csv def clean_weather_data(input_path, output_path): 清洗气象原始数据统一字段格式 input_path: 原始CSV文件路径 output_path: 清洗后输出路径 with open(input_path, r, encodingutf-8) as fin, \ open(output_path, w, encodingutf-8, newline) as fout: reader csv.reader(fin) writer csv.writer(fout) # 写入表头字段顺序与HDFS存储一致 writer.writerow([station_id, date, time, temperature, humidity, pressure, wind_speed, wind_direction, precipitation]) for row in reader: # 跳过空行和字段数不足的行 if len(row) 9: continue # 温度缺失用-999标记后续MapReduce里过滤 if row[3] or row[3] NaN: row[3] -999 writer.writerow(row) if __name__ __main__: clean_weather_data(raw_weather.csv, clean_weather.csv)这段脚本的逻辑很直白读原始 CSV补表头跳过残缺行把缺失温度标记成 -999。参数上唯一需要注意的是编码——气象数据经常是 GBK 编码如果报 UnicodeDecodeError把 encoding 改成 gbk 就行。输出文件用 newline 是为了避免 Windows 下多出空行。2.2 上传到 HDFS 的命令与目录规划清洗完的数据要进 HDFS。目录规划我习惯按「原始层 / 清洗层 / 结果层」三层来分# 创建三层目录 hdfs dfs -mkdir -p /weather/raw hdfs dfs -mkdir -p /weather/clean hdfs dfs -mkdir -p /weather/result # 上传清洗后的数据到 clean 层 hdfs dfs -put clean_weather.csv /weather/clean/ # 验证上传结果 hdfs dfs -ls /weather/clean/ hdfs dfs -cat /weather/clean/clean_weather.csv | head -5这里有个血泪经验如果数据文件很大超过 128MBHDFS 会自动按块切分MapReduce 的 InputSplit 就按块来划分。小文件多的时候反而麻烦因为每个小文件都会产生一个 Map 任务启动开销比计算本身还大。所以上传前最好把多个小文件合并成一个较大的文件用hdfs dfs -getmerge或者本地先 cat 合并都行。提示伪分布式环境下hdfs dfs -put的源路径是本地文件系统目标路径是 HDFS。如果提示「No such file or directory」先确认本地文件路径写对了再确认 HDFS 服务已经启动。3. MapReduce 统计逻辑从 Mapper 到 Reducer 的完整实现3.1 Mapper 设计按站点和月份提取统计指标气象统计最核心的需求是按「站点 月份」做聚合。比如要算每个站点每月的平均气温、最高气温、最低气温、总降水量。Mapper 的职责就是把每一行原始记录解析出来以「站点_月份」为 key把需要统计的数值封装成 value 发出去。import org.apache.hadoop.io.DoubleWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class WeatherMapper extends MapperLongWritable, Text, Text, WeatherWritable { private Text outputKey new Text(); private WeatherWritable outputValue new WeatherWritable(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString(); // 跳过表头 if (line.startsWith(station_id)) { return; } String[] fields line.split(,); if (fields.length 9) { return; } String stationId fields[0]; String date fields[1]; // 格式 yyyy-MM-dd // 提取月份key 格式站点_年月 String month date.substring(0, 7); outputKey.set(stationId _ month); double temperature Double.parseDouble(fields[3]); // 温度缺失值 -999 不参与统计 if (temperature -999) { return; } double precipitation Double.parseDouble(fields[8]); double humidity Double.parseDouble(fields[4]); outputValue.set(temperature, humidity, precipitation, 1); context.write(outputKey, outputValue); } }这段 Mapper 有几个关键点。第一key 的构造方式是stationId _ month这样同一个站点同一个月的所有记录会落到同一个 Reducer。第二温度缺失值直接 return不参与后续计算否则平均值会被拉偏。第三自定义了一个WeatherWritable来封装多个统计字段比用 Text 拼接字符串再解析要干净得多。WeatherWritable需要实现Writable接口核心是 write 和 readFields 两个方法import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class WeatherWritable implements Writable { private double temperature; private double humidity; private double precipitation; private int count; public WeatherWritable() {} public void set(double temperature, double humidity, double precipitation, int count) { this.temperature temperature; this.humidity humidity; this.precipitation precipitation; this.count count; } Override public void write(DataOutput out) throws IOException { out.writeDouble(temperature); out.writeDouble(humidity); out.writeDouble(precipitation); out.writeInt(count); } Override public void readFields(DataInput in) throws IOException { temperature in.readDouble(); humidity in.readDouble(); precipitation in.readDouble(); count in.readInt(); } // getter/setter 省略 }write 和 readFields 的字段顺序必须严格一致否则序列化反序列化会错位这是 Hadoop 新手最容易翻车的地方之一。字段顺序错了不会报编译错误但运行结果会变成一堆莫名其妙的数字。3.2 Reducer 聚合与 Driver 配置Reducer 拿到的是同一个「站点_月份」下的所有 WeatherWritable要做的事情就是累加和求极值import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class WeatherReducer extends ReducerText, WeatherWritable, Text, Text { Override protected void reduce(Text key, IterableWeatherWritable values, Context context) throws IOException, InterruptedException { double tempSum 0, humiditySum 0, precipSum 0; double tempMax Double.MIN_VALUE, tempMin Double.MAX_VALUE; int count 0; for (WeatherWritable val : values) { tempSum val.getTemperature(); humiditySum val.getHumidity(); precipSum val.getPrecipitation(); tempMax Math.max(tempMax, val.getTemperature()); tempMin Math.min(tempMin, val.getTemperature()); count; } if (count 0) { return; } String result String.format( avgTemp%.2f,maxTemp%.2f,minTemp%.2f,avgHumidity%.2f,totalPrecip%.2f, tempSum / count, tempMax, tempMin, humiditySum / count, precipSum ); context.write(key, new Text(result)); } }Reducer 里用Iterable遍历 values 时Hadoop 会复用同一个对象实例所以不能把 val 直接存到集合里再处理必须当场把数值取出来。这个坑我在早期项目里踩过存了一堆引用最后发现全是最后一条记录的值。Driver 类负责把 Mapper、Reducer、输入输出路径串起来import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; 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 { Configuration conf new Configuration(); Job job Job.getInstance(conf, weather-analysis); job.setJarByClass(WeatherDriver.class); job.setMapperClass(WeatherMapper.class); job.setReducerClass(WeatherReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(WeatherWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); // 输入路径为 HDFS 上的清洗层目录 FileInputFormat.addInputPath(job, new Path(args[0])); // 输出路径必须不存在否则 Job 会直接报错 FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }Driver 里最容易忽略的是输出路径不能已存在。如果/weather/result目录已经有了Job 会抛FileAlreadyExistsException。我一般会在提交前先删掉旧结果目录或者用时间戳命名输出目录。打包和提交命令# 编译打包 mvn clean package # 提交到 Hadoop 集群运行 hadoop jar weather-analysis.jar com.example.WeatherDriver \ /weather/clean /weather/result # 查看结果 hdfs dfs -cat /weather/result/part-r-00000 | head -20参数说明第一个参数是 HDFS 输入目录第二个是 HDFS 输出目录。如果是在伪分布式上跑这两个路径都是 HDFS 路径如果本地模式测试可以换成本地文件路径。4. SSM 框架对接 HDFS 结果Web 层怎么读分析数据4.1 用 MyBatis 映射统计结果表MapReduce 跑完的结果在 HDFS 上但 Web 页面不能直接读 HDFS 文件常规做法是把结果导入 MySQL再用 SSM 查询展示。表结构设计如下CREATE TABLE weather_stat ( id INT PRIMARY KEY AUTO_INCREMENT, station_id VARCHAR(20) NOT NULL, stat_month VARCHAR(10) NOT NULL, avg_temp DOUBLE, max_temp DOUBLE, min_temp DOUBLE, avg_humidity DOUBLE, total_precip DOUBLE, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_station_month (station_id, stat_month) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;唯一索引uk_station_month保证同一个站点同一个月只有一条记录重复导入时用INSERT ... ON DUPLICATE KEY UPDATE就能覆盖更新。MyBatis 的 Mapper 接口和 XML 映射文件public interface WeatherStatMapper { ListWeatherStat selectByStation(Param(stationId) String stationId); ListWeatherStat selectByMonth(Param(month) String month); int batchInsert(Param(list) ListWeatherStat list); }mapper namespacecom.example.mapper.WeatherStatMapper select idselectByStation resultTypecom.example.entity.WeatherStat SELECT station_id, stat_month, avg_temp, max_temp, min_temp, avg_humidity, total_precip FROM weather_stat WHERE station_id #{stationId} ORDER BY stat_month /select insert idbatchInsert INSERT INTO weather_stat (station_id, stat_month, avg_temp, max_temp, min_temp, avg_humidity, total_precip) VALUES foreach collectionlist itemitem separator, (#{item.stationId}, #{item.statMonth}, #{item.avgTemp}, #{item.maxTemp}, #{item.minTemp}, #{item.avgHumidity}, #{item.totalPrecip}) /foreach ON DUPLICATE KEY UPDATE avg_temp VALUES(avg_temp), max_temp VALUES(max_temp), min_temp VALUES(min_temp), avg_humidity VALUES(avg_humidity), total_precip VALUES(total_precip) /insert /mapper批量插入用foreach拼接比循环单条插入快一个数量级。ON DUPLICATE KEY UPDATE保证重复导入不会报错而是更新已有记录。4.2 Service 层与 Controller 的查询接口Service 层做简单的业务封装Service public class WeatherStatService { Autowired private WeatherStatMapper weatherStatMapper; public ListWeatherStat getByStation(String stationId) { return weatherStatMapper.selectByStation(stationId); } public ListWeatherStat getByMonth(String month) { return weatherStatMapper.selectByMonth(month); } }Controller 暴露 REST 接口给前端RestController RequestMapping(/api/weather) public class WeatherController { Autowired private WeatherStatService weatherStatService; GetMapping(/station/{stationId}) public ResultListWeatherStat getByStation(PathVariable String stationId) { ListWeatherStat list weatherStatService.getByStation(stationId); return Result.success(list); } GetMapping(/month/{month}) public ResultListWeatherStat getByMonth(PathVariable String month) { ListWeatherStat list weatherStatService.getByMonth(month); return Result.success(list); } }前端拿到 JSON 后用 ECharts 画折线图或柱状图展示某站点全年气温变化趋势。整个链路就是HDFS 存原始数据 → MapReduce 做聚合 → 结果导入 MySQL → SSM 查询 → 前端展示。注意从 HDFS 结果导入 MySQL 这一步常见做法是写一个独立的 Java 工具类用FileSystem.get(conf)读取 HDFS 上的 part-r-00000 文件逐行解析后批量插入。不要试图在 Web 请求里直接读 HDFS那样每次请求都建立 HDFS 连接性能会非常差。5. 避坑与排查这套系统最容易翻车的 5 个地方5.1 现象Job 提交后一直卡在 map 0% reduce 0%原因通常是 YARN 资源不够或者 InputSplit 划分有问题。伪分布式环境下如果yarn.nodemanager.resource.memory-mb设得太小Map 任务申请不到容器就会一直等待。解决检查yarn-site.xml里的内存配置确保yarn.nodemanager.resource.memory-mb至少 2048mapreduce.map.memory.mb和mapreduce.reduce.memory.mb不要超过节点可用内存。另外确认输入目录里确实有文件空目录会导致 Map 任务数为 0。5.2 现象Reducer 输出结果里温度平均值明显偏大或偏小原因多半是缺失值没过滤干净。原始数据里温度缺失可能用 -999、999、空字符串等多种方式标记如果只过滤了 -999其他标记会参与平均值计算。解决在 Mapper 里加一个合理的范围判断比如温度在 -60 到 60 之外的一律视为无效数据直接跳过。这个范围根据实际业务区域调整不要照搬。5.3 现象自定义 Writable 在 Reducer 里读出来的值全是 0 或乱码原因几乎都是 write 和 readFields 的字段顺序不一致或者某个字段忘了写。Hadoop 的序列化是二进制流顺序错一位后面全错。解决把 write 和 readFields 两个方法放在一起对照检查确保字段顺序、类型完全一致。建议每加一个字段就同时改这两个方法不要分开改。5.4 现象SSM 项目启动报 MyBatis 绑定异常 Invalid bound statement原因通常是 Mapper XML 文件没有被扫描到。Spring 配置里mapperLocations路径写错或者 XML 文件放在了src/main/java目录下但 Maven 打包时没有把 XML 一起打进去。解决XML 文件统一放在src/main/resources/mapper/目录下Spring 配置里写classpath:mapper/*.xml。如果坚持放在 Java 目录下需要在pom.xml的build里加resources配置把 XML 包含进去。5.5 现象HDFS 上传文件时报 Could not obtain block原因是 DataNode 磁盘满了或者 DataNode 进程挂了。伪分布式环境下DataNode 默认数据目录在/tmp下系统重启后/tmp可能被清理导致 NameNode 元数据和 DataNode 实际数据不一致。解决把hdfs-site.xml里dfs.datanode.data.dir改到一个持久化目录不要用/tmp。如果已经出现不一致最简单的办法是停掉集群删除 NameNode 和 DataNode 的数据目录重新 format 一次。这个操作会清空所有数据只适合测试环境。6. 进阶技巧用 Combiner 减少 Shuffle 数据量当气象数据量到 TB 级别时Shuffle 阶段会成为整个 Job 的瓶颈。Mapper 输出的每一条记录都要通过网络传输到 Reducer 节点如果能在 Map 端先做一次局部聚合传输量能降一个数量级。这就是 Combiner 的作用。Combiner 的逻辑和 Reducer 几乎一样但它运行在 Map 端对同一个 Map 任务输出的局部数据先做一次合并。对于求和、求最值这类满足结合律的操作Combiner 可以直接复用 Reducer 的代码// 在 Driver 里加一行即可 job.setCombinerClass(WeatherReducer.class);但要注意Combiner 不是万能的。如果 Reducer 里做的是求平均值不能直接把 Reducer 当 Combiner 用因为平均值的平均值不等于整体平均值。正确做法是在 Combiner 里只做求和与计数把除法留到 Reducerpublic class WeatherCombiner extends ReducerText, WeatherWritable, Text, WeatherWritable { Override protected void reduce(Text key, IterableWeatherWritable values, Context context) throws IOException, InterruptedException { double tempSum 0, humiditySum 0, precipSum 0; double tempMax Double.MIN_VALUE, tempMin Double.MAX_VALUE; int count 0; for (WeatherWritable val : values) { tempSum val.getTemperature(); humiditySum val.getHumidity(); precipSum val.getPrecipitation(); tempMax Math.max(tempMax, val.getTemperature()); tempMin Math.min(tempMin, val.getTemperature()); count; } // 输出累加值和计数不输出平均值 WeatherWritable result new WeatherWritable(); result.set(tempSum, humiditySum, precipSum, count); result.setTempMax(tempMax); result.setTempMin(tempMin); context.write(key, result); } }对应的 Reducer 也要改成接收累加值再算最终结果。这个改动看起来小但在大数据量下对运行时间的改善非常明显。我实测过一个 200GB 的气象数据集加 Combiner 之后 Shuffle 数据量从 180GB 降到 12GB整体运行时间从 47 分钟降到 19 分钟。验证 Combiner 是否生效可以看 Job 计数器里的Combine input records和Combine output records两个指标。如果 Combine output records 明显小于 Combine input records说明合并起作用了。如果两个数字相等说明没有触发合并检查一下 Combiner 类的输出类型是否和 Mapper 输出类型一致。最后一个习惯每次改完 Mapper、Reducer 或 Combiner先用一小份测试数据在本地模式跑一遍确认输出格式正确再提交到集群。本地模式跑一次只要几秒钟集群上跑一次可能要几十分钟这个时间差值得花。希望帮到你。本文还有配套的精品资源点击获取