气象大数据案例项目(求各气象站的平均气温)
- 一、项目需求
- 二、数据格式
- 三、项目开发
- 3.1 在windows 进行开发
- 3.2 运行结果
- 3.3 对项目打包
一、项目需求
现在有一份来自美国国家海洋和大气管理局的数据集,里面包含近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