气象大数据案例项目（求各气象站的平均气温）

一、项目需求

现在有一份来自美国国家海洋和大气管理局的数据集，里面包含近30年每个气象站、每小时的天气预报数据，每个报告的文件大小大约15M。一共有10个气象站，每个报告文件的名字包含气象站ID，每条记录包含气温、风向、天气状况等多个字段信息。现在要求统计美国各气象站30年平均气温。

二、数据格式

一共10份气象站的数据

文档里面的数据格式，注意 -9999 说明数据缺失

三、项目开发

3.1 在windows 进行开发

• 引入 Hadoop 依赖

<dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>2.10.2</version>
</dependency>

• 开发脚本

package com.feifei.mapreduce;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;import java.io.IOException;public class WeatherAnalysis {public static class MyMapper extends Mapper<Object, Text, Text, IntWritable> {@Overrideprotected void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {String line = value.toString();int temperature = Integer.parseInt(line.substring(14, 19).trim());if(temperature != -9999){FileSplit fileSplit = (FileSplit) context.getInputSplit();String id = fileSplit.getPath().getName().substring(5, 10);context.write(new Text(id), new IntWritable(temperature));}}}public static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {private IntWritable mean = new IntWritable();@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {int sum = 0;int count = 0;for (IntWritable val : values) {sum += val.get();}mean.set(sum / count);context.write(key, mean);}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(WeatherAnalysis.class);job.setJobName("WeatherAnalysis");job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutputFormat.class);FileInputFormat.addInputPath(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));job.setMapperClass(WeatherAnalysis.MyMapper.class);job.setReducerClass(WeatherAnalysis.MyReducer.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);job.waitForCompletion(true);}
}

• 设置入参和保存路径

3.2 运行结果

3.3 对项目打包

 mvn clean package