MapReduce程序之求一年中的最高温度和最低温度
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了MapReduce程序之求一年中的最高温度和最低温度相关的知识,希望对你有一定的参考价值。
[TOC]MapReduce程序之求一年中的最高温度和最低温度
前言
看过《Hadoop权威指南》的同学都知道,关于MapReduce的第一个入门的例子就是统计全球气温,书上的例子是使用了全部的数据来作为统计,但实际上只需要拿某一年的数据来作为测试也就OK了,所以下面写的程序用的数据是某一年的气温数据。
数据获取与说明
可以在下面的网址中下载到全部的数据:
ftp://ftp.ncdc.noaa.gov/pub/data/gsod/
同时这个网址也有提供关于数据每个字段的说明,也就是readme.txt文件,因为我们关注的是气温的最大与最小值,所以只需要查看相关的说明即可,其关于气温最值说明如下:
MAX 103-108 Real Maximum temperature reported during the
day in Fahrenheit to tenths--time of max
temp report varies by country and
region, so this will sometimes not be
the max for the calendar day. Missing =
9999.9
MIN 111-116 Real Minimum temperature reported during the
day in Fahrenheit to tenths--time of min
temp report varies by country and
region, so this will sometimes not be
the min for the calendar day. Missing =
9999.9
也就是说,每行的第103-108个字符为当天的最高气温,第111-116个字符为当天的最低气温,基于此就可以写出我们的MapReduce程序了。
程序思路
/**
数据源:ftp://ftp.ncdc.noaa.gov/pub/data/gsod/
求出一年中的最高温度和最低温度。
* MR应用程序
*
* Map<k1, v1, k2, v2>
* 第一步:确定map的类型参数
* k1, v1是map函数的输入参数
* k2, v2是map函数的输出参数
* 对于普通的文本文件的每一行的起始偏移量就是k1,---->Long(LongWritable)
* 对于普通的文本文件,v2就是其中的一行数据,是k1所对应的一行数据,---->String(Text)
* k2, v2
* k2就是拆分后的单词,---->String(Text)
* v2就是温度---->double(DoubleWritable)
* 第二步:编写一个类继承Mapper
* 复写其中的map函数
* Reduce<k2, v2s, k3, v3>
* 第一步:确定reduce的类型
* k2, v2s是reduce函数的输入参数
* k3, v3是reduce函数的输出参数
* k2 --->Text
* v2s ---->Iterable<DoubleWritable>
*
* k3 聚合之后的单词---->Text
* v3 聚合之后的单词对应的次数--->DoubleWritable
第二步:编写一个类继承Reducer
* 复写其中的reduce函数
*
*
* 第三步:编写完map和reduce之后,将二者通过驱动程序组装起来,进行执行
*
*
* mr的执行的方式:
* yarn/hadoop jar jar的路径 全类名 参数
*/
MapReduce程序
根据程序思路和数据格式,写出的MapReduce程序如下:
package com.uplooking.bigdata.mr.weather;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
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 WeatherJob {
public static void main(String[] args) throws Exception {
if (args == null || args.length < 2) {
System.err.println("Parameter Errors! Usages:<inputpath> <outputpath>");
System.exit(-1);
}
Path inputPath = new Path(args[0]);
Path outputPath = new Path(args[1]);
Configuration conf = new Configuration();
String jobName = WeatherJob.class.getSimpleName();
Job job = Job.getInstance(conf, jobName);
//设置job运行的jar
job.setJarByClass(WeatherJob.class);
//设置整个程序的输入
FileInputFormat.setInputPaths(job, inputPath);
job.setInputFormatClass(TextInputFormat.class);//就是设置如何将输入文件解析成一行一行内容的解析类
//设置mapper
job.setMapperClass(WeatherMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(DoubleWritable.class);
//设置整个程序的输出
// outputpath.getFileSystem(conf).delete(outputpath, true);//如果当前输出目录存在,删除之,以避免.FileAlreadyExistsException
FileOutputFormat.setOutputPath(job, outputPath);
job.setOutputFormatClass(TextOutputFormat.class);
//设置reducer
job.setReducerClass(WeatherReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(DoubleWritable.class);
//指定程序有几个reducer去运行
job.setNumReduceTasks(1);
//提交程序
job.waitForCompletion(true);
}
public static class WeatherMapper extends Mapper<LongWritable, Text, Text, DoubleWritable> {
@Override
protected void map(LongWritable k1, Text v1, Context context) throws IOException, InterruptedException {
String line = v1.toString();
Double max = null;
Double min = null;
try {
// 获取一行中的气温MAX值
max = Double.parseDouble(line.substring(103, 108));
// 获取一行中的气温MIN值
min = Double.parseDouble(line.substring(111, 116));
} catch (NumberFormatException e) {
// 如果出现异常,则当前的这一个map task不执行,直接返回
return;
}
// 写到context中
context.write(new Text("MAX"), new DoubleWritable(max));
context.write(new Text("MIN"), new DoubleWritable(min));
}
}
public static class WeatherReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {
@Override
protected void reduce(Text k2, Iterable<DoubleWritable> v2s, Context context) throws IOException, InterruptedException {
// 先预定义最大和最小气温值
double max = Double.MIN_VALUE;
double min = Double.MAX_VALUE;
// 得到迭代列表中的气温最大值和最小值
if ("MAX".equals(k2.toString())) {
for (DoubleWritable v2 : v2s) {
double tmp = v2.get();
if (tmp > max) {
max = tmp;
}
}
} else {
for (DoubleWritable v2 : v2s) {
double tmp = v2.get();
if (tmp < min) {
min = tmp;
}
}
}
// 将结果写入到context中
context.write(k2, "MAX".equals(k2.toString()) ? new DoubleWritable(max) : new DoubleWritable(min));
}
}
}
测试
注意,上面的程序是需要读取命令行的参数输入的,可以在本地的环境执行,也可以打包成一个jar包上传到Hadoop环境的Linux服务器上执行,这里,我使用的是本地环境(我的操作系统是Mac OS),输入的参数如下:
/Users/yeyonghao/data/input/010010-99999-2015.op /Users/yeyonghao/data/output/mr/weather/w-1
执行程序,输出结果如下:
...省略部分输出...
2018-03-06 11:23:14,915 [main] [org.apache.hadoop.mapreduce.Job] [INFO] - Job job_local2102200632_0001 running in uber mode : false
2018-03-06 11:23:14,917 [main] [org.apache.hadoop.mapreduce.Job] [INFO] - map 100% reduce 100%
2018-03-06 11:23:14,918 [main] [org.apache.hadoop.mapreduce.Job] [INFO] - Job job_local2102200632_0001 completed successfully
2018-03-06 11:23:14,927 [main] [org.apache.hadoop.mapreduce.Job] [INFO] - Counters: 30
...省略部分输出...
MapReduce程序执行成功后,再查看输出目录中的输出结果:
[email protected]:~/data/output/mr/weather/w-1$ cat part-r-00000
MAX 57.6
MIN 6.3
可以看到,已经可以正确统计出最大气温和最小气温值,说明我们的MapReduce程序没有问题。
以上是关于MapReduce程序之求一年中的最高温度和最低温度的主要内容,如果未能解决你的问题,请参考以下文章