Hadoop MapReduce实战：用Java代码手把手教你清洗气象数据（附完整代码）

18790970257

378人浏览 · 2026-06-06 10:58:33

18790970257 · 2026-06-06 10:58:33 发布

Hadoop MapReduce实战：用Java代码手把手教你清洗气象数据（附完整代码）

气象数据作为典型的时间序列数据，其清洗过程往往面临字段多、规则复杂、数据量大等挑战。本文将基于真实气象数据集，通过一个完整的MapReduce项目，演示如何从原始数据中提取有效信息，并解决实际开发中的典型问题。

1. 项目环境与数据准备

在开始编码前，我们需要明确数据结构和清洗目标。原始数据包含两个关键文件：

a.txt ：主数据文件，空格分隔，包含12个字段
sky.txt ：天气代码映射文件，逗号分隔

典型的气象数据行如下：

2005 01 01 16 -6 -28 10157 260 31 8 0 -9999

字段说明表：

位置	字段名	数据类型	有效范围
0	年	字符串	-
1	月	字符串	-
2	日	字符串	-
3	小时	字符串	-
4	温度	整型	[-40,50]
5	湿度	整型	[0,100]
6	气压	整型	>0
7	风向	整型	[0,360]
8	风速	整型	≥0
9	天气代码	整型	[0,10]
10	1h降雨量	字符串	-
11	6h降雨量	字符串	-

提示：实际项目中建议使用Parquet或ORC列式存储格式，可显著提升处理效率

2. 核心Java类设计与实现

2.1 Weather数据封装类

作为MapReduce的Value对象，需要实现WritableComparable接口：

public class Weather implements WritableComparable<Weather> {
    private String year;
    private String month;
    // 其他字段...
    
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(year);
        out.writeUTF(month);
        // 其他字段序列化...
    }
    
    @Override
    public int compareTo(Weather o) {
        // 实现多级排序规则
        int cmp = this.year.compareTo(o.year);
        if (cmp == 0) {
            cmp = this.month.compareTo(o.month);
            if (cmp == 0) {
                cmp = this.day.compareTo(o.day);
                if (cmp == 0) {
                    cmp = Integer.compare(this.temperature, o.temperature);
                    // 继续其他排序条件...
                }
            }
        }
        return cmp;
    }
}

2.2 Mapper类实现数据清洗

Mapper需要完成三项核心工作：

加载天气代码映射表
过滤无效数据
转换数据格式

public class WeatherMapper extends Mapper<LongWritable, Text, Weather, NullWritable> {
    private Map<String, String> skyConditionMap = new HashMap<>();
    
    @Override
    protected void setup(Context context) throws IOException {
        // 从分布式缓存读取sky.txt
        Path[] cacheFiles = DistributedCache.getLocalCacheFiles(context.getConfiguration());
        try (BufferedReader reader = new BufferedReader(new FileReader(cacheFiles[0].toString()))) {
            String line;
            while ((line = reader.readLine()) != null) {
                String[] parts = line.split(",");
                skyConditionMap.put(parts[0], parts[1]);
            }
        }
    }
    
    @Override
    protected void map(LongWritable key, Text value, Context context) 
        throws IOException, InterruptedException {
        
        String[] fields = value.toString().split("\\s+");
        if (fields.length != 12) return;
        
        // 数据有效性校验
        int temperature = Integer.parseInt(fields[4]);
        if (temperature < -40 || temperature > 50) return;
        
        // 其他字段校验...
        
        // 天气代码转换
        String skyCondition = skyConditionMap.getOrDefault(fields[9], "未知");
        
        Weather weather = new Weather(fields[0], fields[1], /* 其他字段 */);
        context.write(weather, NullWritable.get());
    }
}

3. 高级功能实现

3.1 自定义分区器

按年份分区可优化数据分布：

public class YearPartitioner extends Partitioner<Weather, NullWritable> {
    @Override
    public int getPartition(Weather key, NullWritable value, int numPartitions) {
        // 简单哈希分区示例
        return (key.getYear().hashCode() & Integer.MAX_VALUE) % numPartitions;
    }
}

3.2 Reducer优化处理

虽然本例可直接输出Mapper结果，但复杂场景可能需要聚合：

public class WeatherReducer extends Reducer<Weather, NullWritable, Weather, NullWritable> {
    @Override
    protected void reduce(Weather key, Iterable<NullWritable> values, Context context) 
        throws IOException, InterruptedException {
        
        // 可在此添加聚合逻辑
        context.write(key, NullWritable.get());
    }
}

4. 作业配置与执行

完整的作业驱动类配置：

public class WeatherJob extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf(), "Weather Data Cleaning");
        job.setJarByClass(WeatherJob.class);
        
        // 输入输出路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        // 添加分布式缓存
        job.addCacheFile(new Path("sky.txt").toUri());
        
        // Mapper配置
        job.setMapperClass(WeatherMapper.class);
        job.setMapOutputKeyClass(Weather.class);
        job.setMapOutputValueClass(NullWritable.class);
        
        // Reducer配置
        job.setReducerClass(WeatherReducer.class);
        job.setOutputKeyClass(Weather.class);
        job.setOutputValueClass(NullWritable.class);
        
        // 自定义分区
        job.setPartitionerClass(YearPartitioner.class);
        job.setNumReduceTasks(3);  // 根据实际数据量调整
        
        return job.waitForCompletion(true) ? 0 : 1;
    }
    
    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new WeatherJob(), args);
        System.exit(exitCode);
    }
}

5. 实战问题排查指南

开发过程中常见问题及解决方案：

数据倾斜问题
- 现象：个别Reducer处理数据量远大于其他
- 解决：优化分区策略，或使用Combiner预聚合
内存溢出
- 现象：java.lang.OutOfMemoryError
- 解决：调整mapreduce.map.memory.mb等参数
字段解析异常
- 现象：NumberFormatException等
- 解决：增加健壮性检查代码

// 改进后的字段解析示例
try {
    temperature = Integer.parseInt(field);
} catch (NumberFormatException e) {
    context.getCounter("Data Quality", "Invalid Temperature").increment(1);
    return;
}