简介这是一套面向大数据与气象交叉领域学习者的Hadoop实战代码包围绕分布式计算处理气象数据展开适合已掌握Java基础、希望理解HDFS与MapReduce实际应用的学生或开发者。资源以MapReduce为核心涵盖数据清洗、分组、聚合等环节可计算平均气温、最高最低温度、降雨量等指标并借助SSM框架将分析结果通过Web界面展示涉及Spring Boot、Spring MVC与MyBatis的整合思路。压缩包共562个文件约34.88MB包含124个js、113个png、78个jpg、69个css等前端资源60个xml与28个jar支撑工程配置另有15个java、26个class、10个jsp及1个war体现完整项目结构并附sql脚本便于数据落地。目前已有10587人学习下载读者可从中获取从分布式存储、并行计算到结果可视化的完整链路参考理解气象数据预处理、统计分析与Web呈现的衔接方式为课程设计或大数据入门项目提供可复用的代码骨架与排错思路。1. 从一份气象数据代码包说起Hadoop 处理日均温的真实门槛很多人第一次拿到「Hadoop分析气象数据完整版代码」这类资源脑子里想的是解压、编译、跑通、出图。但真正动手才会发现卡住你的不是 MapReduce 本身而是气象数据那种「半结构化、按行定长、字段靠位置切」的原始格式。这份代码包解决的核心问题是把气象站历史观测记录里的日均温、最高温、最低温用 Hadoop MapReduce 跑出年度极值和月度均值。它适合两类人一是正在学 Hadoop 但找不到真实数据练手的开发者二是手里有气象观测数据、想用分布式思路做批量统计的从业者。代码本身不复杂难的是数据格式解析和分区逻辑这两块踩坑最多。2. 气象数据格式拆解从定长记录到 MapReduce 输入2.1 气象原始数据的字段布局常见的气象观测数据来自地面站逐日观测格式通常是每行一条记录字段按固定宽度排列。以典型的 GHCN 风格数据为例一行里包含站号、年份、月份、要素类型、若干天数值。要素类型用四个字符标识比如 TMAX 表示日最高温、TMIN 表示日最低温、PRCP 表示降水量。每个日值占 5 个字符前 4 位是数值最后一位是标志位表示该值是否有效、是否估算、是否缺测。这种格式对 MapReduce 来说有个天然优势每一行独立不需要跨行关联天然适合按行切分的 InputFormat。但劣势也很明显——字段没有分隔符不能直接用逗号或制表符切。你得按字符位置去截取一旦位置算错整个统计结果就是错的。我见过有人直接把整行按空格 split结果站号里带空格的行全乱了。2.2 为什么选 MapReduce 而不是 Spark这份代码包用的是 MapReduce不是 Spark。有人会问现在 Spark 不是更快吗从技术选型角度气象数据年度统计这个场景有几个特点数据量大但计算逻辑简单、不需要迭代计算、不需要低延迟交互。MapReduce 的磁盘落盘机制在这种「一次扫描、批量聚合」的场景下反而更稳尤其是集群资源有限、内存不够大的时候Spark 容易 OOMMapReduce 慢慢跑总能出结果。另外这份代码包的定位是教学和练手MapReduce 的编程模型更直观地体现了「分而治之」的思想。Mapper 负责解析每条记录、提取有效温度值Reducer 负责按站号和年份聚合。整个流程清晰适合理解分布式计算的基本范式。如果你已经熟悉 MapReduce想迁移到 Spark核心逻辑不变只是 API 换一下。2.3 输入格式的代码实现代码包里自定义了一个 InputFormat继承自 FileInputFormatRecordReader 按行读取每行作为一个完整记录传给 Mapper。这样做的好处是不依赖默认的 TextInputFormat 对行长的限制气象数据里偶尔有超长行比如某天补录了大量数据默认的 LineRecordReader 在极端情况下会截断。// 自定义 InputFormat确保每行完整读取 public class WeatherInputFormat extends FileInputFormatLongWritable, Text { Override public RecordReaderLongWritable, Text createRecordReader( InputSplit split, TaskAttemptContext context) { return new WeatherLineRecordReader(); } } // RecordReader 实现按行读取不设行长上限 public class WeatherLineRecordReader extends RecordReaderLongWritable, Text { private LineReader lineReader; private LongWritable key new LongWritable(); private Text value new Text(); private long pos 0; Override public void initialize(InputSplit split, TaskAttemptContext context) throws IOException { FileSplit fileSplit (FileSplit) split; Path path fileSplit.getPath(); FileSystem fs path.getFileSystem(context.getConfiguration()); FSDataInputStream in fs.open(path); // 关键参数缓冲区大小默认 64KB气象数据行较短64KB 足够 lineReader new LineReader(in, context.getConfiguration()); // 跳过 split 起始位置之前的内容避免重复读取 if (fileSplit.getStart() ! 0) { in.seek(fileSplit.getStart() - 1); lineReader.readLine(new Text()); } } Override public boolean nextKeyValue() throws IOException { // 每次读取一行返回 true 表示还有数据 int newSize lineReader.readLine(value); if (newSize 0) { return false; } key.set(pos); pos newSize; return true; } // 省略 getCurrentKey/getCurrentValue/close 等标准方法 }这段代码的关键点在于initialize里的 seek 逻辑。Hadoop 会把大文件切成多个 split每个 split 从一个字节偏移量开始。如果直接从这个偏移量读可能读到半行。所以要先回退一个字节读掉不完整的那行再从下一行开始正式处理。这个细节在默认的 LineRecordReader 里已经处理了但自定义 RecordReader 时很容易漏掉漏掉就会导致数据重复或丢失。3. Mapper 与 Reducer 核心逻辑温度提取和年度聚合3.1 Mapper 的解析逻辑与容错Mapper 的任务很明确读一行气象记录判断要素类型是不是 TMAX 或 TMIN如果是提取年份、站号和每天的温度值输出站号年份, 温度的键值对。但实际数据里有一堆脏情况缺测值用 -9999 表示、标志位标记为无效的值要跳过、某些行要素类型不在关注范围内直接丢弃。public class TemperatureMapper extends MapperLongWritable, Text, Text, IntWritable { private Text outKey new Text(); private IntWritable outValue new IntWritable(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString(); // 气象数据行长度至少 21 字符站号11年份4月份2要素4 if (line.length() 21) { return; // 太短的行直接跳过可能是文件头或损坏行 } String stationId line.substring(0, 11).trim(); String year line.substring(11, 15).trim(); String element line.substring(17, 21).trim(); // 只处理最高温和最低温 if (!TMAX.equals(element) !TMIN.equals(element)) { return; } // 从第 21 位开始每 5 个字符一个日值 for (int i 21; i 5 line.length(); i 5) { String tempStr line.substring(i, i 4).trim(); String flag line.substring(i 4, i 5); // 标志位为 9 表示缺测直接跳过 if (9.equals(flag)) { continue; } try { int temp Integer.parseInt(tempStr); // 缺测值 -9999 跳过 if (temp -9999) { continue; } outKey.set(stationId _ year); outValue.set(temp); context.write(outKey, outValue); } catch (NumberFormatException e) { // 解析失败的行不中断任务记录日志后跳过 System.err.println(解析失败: line.substring(i, i 5)); } } } }这里有几个参数需要根据实际数据调整。substring(0, 11)里的 11 是站号宽度不同数据集可能不同GHCN 是 11 位但有些国内数据集是 5 位站号。substring(17, 21)里的 17 是要素类型起始位置这个偏移量取决于年份和月份字段的宽度。如果数据格式有差异这两个数字必须改否则提取出来的全是乱码。3.2 Reducer 的聚合策略Reducer 收到的是同一个站号年份下的所有温度值需要算出最高温、最低温和平均温。这里有个容易翻车的地方TMAX 和 TMIN 在 Mapper 阶段没有区分都输出成了同一个键。Reducer 拿到的一堆值里既有最高温也有最低温直接取 max 和 min 会混淆。代码包里的做法是在 Mapper 输出时把要素类型拼进 key比如站号_年份_TMAX这样 Reducer 就能分开处理。但这样会导致 Reduce 任务数翻倍。另一种做法是在 Reducer 里根据值的范围做二次判断但气象数据里最高温和最低温的范围有重叠不可靠。所以推荐第一种key 里带要素类型。public class TemperatureReducer extends ReducerText, IntWritable, Text, Text { Override protected void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException { int maxTemp Integer.MIN_VALUE; int minTemp Integer.MAX_VALUE; long sum 0; int count 0; for (IntWritable val : values) { int temp val.get(); maxTemp Math.max(maxTemp, temp); minTemp Math.min(minTemp, temp); sum temp; count; } if (count 0) { double avgTemp (double) sum / count; // 输出格式站号_年份 最高温/最低温/平均温 String result maxTemp / minTemp / String.format(%.1f, avgTemp); context.write(key, new Text(result)); } } }Reducer 里的sum用 long 而不是 int是因为一个站号一年的温度值累加可能超过 int 范围。虽然单个温度值不大但累加几千次后 int 有溢出风险。这个细节在数据量小的时候看不出来数据量一大就出负数属于典型的「小数据测试通过、大数据翻车」问题。3.3 Driver 配置与分区数设置Driver 类负责组装 Job设置 Mapper、Reducer、输入输出路径。这里最关键的参数是 Reduce 任务数。默认是 1所有数据涌到一个 Reducer数据量大的时候慢得离谱。一般设置为集群可用 Reduce slot 数的 0.9 倍左右或者根据输出文件大小目标来定。public class WeatherDriver { public static void main(String[] args) throws Exception { Configuration conf new Configuration(); Job job Job.getInstance(conf, weather-annual-stats); job.setJarByClass(WeatherDriver.class); job.setInputFormatClass(WeatherInputFormat.class); job.setMapperClass(TemperatureMapper.class); job.setReducerClass(TemperatureReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); // 关键参数Reduce 任务数根据集群规模调整 job.setNumReduceTasks(4); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }setNumReduceTasks(4)这个值不是随便写的。如果集群有 4 个 Reduce slot设 4 刚好跑满设 8 会导致部分任务排队等待反而增加总时间。另外Reduce 数决定了最终输出文件的数量设 4 就会生成 4 个 part-r-00000 到 part-r-00003。如果后续要合并结果文件数太多也麻烦。4. 避坑与排查气象数据 MapReduce 的五个血泪教训4.1 现象任务跑完输出为空日志无报错原因Mapper 里的字段偏移量写错了。比如站号实际是 5 位代码里按 11 位截取substring(0, 11)把年份和月份也截进去了导致element字段取到的是空字符串或乱码所有记录都被if (!TMAX.equals(element))过滤掉。解决先用head -5看原始数据的前几行数清楚每个字段的起始位置和宽度。不要凭记忆写偏移量不同数据集格式差异很大。可以在 Mapper 里加一行context.getCounter(DEBUG, TOTAL_LINES).increment(1)跑完后看计数器确认 Mapper 到底读了多少行、过滤了多少行。4.2 现象Reduce 阶段卡在 99%迟迟不结束原因数据倾斜。某个站号的数据量特别大或者某个年份的记录异常多导致一个 Reducer 处理了大部分数据其他 Reducer 早就跑完了就等这一个。气象数据里这种情况常见于热门站点或特殊年份。解决在 Driver 里设置自定义 Partitioner把数据均匀打散。或者对 key 加随机前缀先做一次局部聚合再去掉前缀做全局聚合。代码包里用的是默认 HashPartitioner如果数据倾斜严重需要换成 RangePartitioner 或自定义实现。4.3 现象输出结果里温度值全是 -9999原因缺测值判断逻辑写反了。代码里写的是if (temp -9999) continue;但如果数据里的缺测值不是 -9999 而是 9999 或其他值这个判断就失效了。另外有些数据集的缺测标志位不是 9而是其他字符。解决先统计一下数据里所有出现的标志位字符用cut命令截取标志位列sort | uniq -c看分布。确认缺测标志到底是哪个字符再改代码。不要假设所有数据集都用同一套缺测约定。4.4 现象本地模式跑得通集群上报 ClassNotFoundException原因自定义的 InputFormat 和 RecordReader 类没有打包进 Job 的 jar 里。本地模式跑的时候类在 classpath 下直接加载提交到集群后只有 job jar 里的类才能被各节点加载。解决确保job.setJarByClass(WeatherDriver.class)指向的类所在的 jar 包含了所有自定义类。用jar tf your-job.jar | grep WeatherInputFormat确认一下。如果用的是 Maven检查 pom 里的打包配置确保自定义类没有被排除。4.5 现象输出文件数量太多后续处理麻烦原因Reduce 任务数设得太大或者没有设置输出合并。每个 Reducer 生成一个 part 文件Reduce 数设成 100 就生成 100 个文件。解决根据最终数据量调整 Reduce 数。如果输出结果不大设 1 个 Reducer 就够了虽然慢一点但文件少。或者用FileOutputFormat.setCompressOutput开启压缩减少磁盘占用。另外Hadoop 的getmerge命令可以把多个 part 文件合并成一个但要注意合并后的文件可能很大后续处理要相应调整。5. 进阶技巧用 Combiner 和自定义计数器提升效率5.1 Combiner 的正确用法与边界Combiner 的本质是在 Map 端做一次本地 Reduce减少网络传输。对于温度统计这种场景Combiner 可以直接复用 Reducer 的逻辑因为求最大值、最小值、累加和都是可结合的操作。但要注意Combiner 的输出类型必须和 Mapper 的输出类型一致否则会报类型转换错误。// 在 Driver 里设置 Combiner job.setCombinerClass(TemperatureReducer.class);这行代码看起来简单但有个前提Reducer 的输入和输出类型必须一致。TemperatureReducer 的输入是Text, IntWritable输出是Text, Text类型不一致不能直接当 Combiner 用。需要单独写一个 Combiner输出Text, IntWritable把最大值、最小值、累加和打包成一个自定义 Writable 传递。代码包里没有用 Combiner因为教学场景下数据量不大网络传输不是瓶颈。但如果你的数据量到了 TB 级Combiner 能减少 60% 以上的网络传输效果非常明显。5.2 自定义计数器监控数据质量Hadoop 的计数器是个被低估的调试工具。除了内置的 Map input records、Reduce output records你可以自定义计数器来统计脏数据比例。比如统计缺测值数量、解析失败行数、超出合理范围的温度值数量。// 在 Mapper 里定义计数器 enum WeatherStats { MISSING_VALUE, // 缺测值数量 PARSE_ERROR, // 解析失败数量 OUT_OF_RANGE, // 超出合理范围-60~60度 VALID_RECORD // 有效记录数 } // 在 map 方法里使用 if (9.equals(flag)) { context.getCounter(WeatherStats.MISSING_VALUE).increment(1); continue; } if (temp -60 || temp 60) { context.getCounter(WeatherStats.OUT_OF_RANGE).increment(1); continue; } context.getCounter(WeatherStats.VALID_RECORD).increment(1);跑完任务后在控制台输出里能看到这些计数器的值。如果 MISSING_VALUE 占比超过 30%说明数据质量有问题可能需要换数据源或调整清洗策略。如果 PARSE_ERROR 数量不为零说明字段偏移量还有问题需要回去检查。5.3 输出结果的验证方法MapReduce 跑完后不要直接拿结果去用。先做几层验证第一用hdfs dfs -cat output/part-r-00000 | head -20看前几行确认格式正确、温度值在合理范围内。第二随机挑一个站号用grep在原始数据里找到该站的所有记录手工算一遍年度极值和 MapReduce 输出对比。第三统计输出行数和原始数据里的唯一站号年份组合数对比确认没有漏算。我一般会写一个简单的 Python 脚本做交叉验证从原始数据里抽样 100 个站号用 pandas 算一遍再和 Hadoop 输出对比。如果两者一致说明逻辑没问题如果不一致差异点就是排查方向。这个习惯帮我抓过好几次字段偏移量错误和缺测值判断遗漏。从那以后我每次跑气象数据统计都强制走一遍「计数器检查 抽样交叉验证」的流程不管数据量大小。希望帮到你。本文还有配套的精品资源点击获取