Java 实现MapReduce函数

明白了MapReduce程序的工作原理之后，下一步就是写代码来实现它。我们需要三样东西：一个map函数、一个reduce函数和一些用来运行作业的代码。map函数由Mapper类来表示，后者声明一个map()虚方法。范例2-3显示了我们的map函数实现。

范例2-3 查找最高气温的Mapper类

Import java.Io.IOException;
import org.apahce.hadoop.io.IntWritable;
import org.apahce.hadoop.io.LongWritable;
import org.apahce.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;public class MaxTemperatureMapper extends MapReduceBase implements Mapper<LongWritable,Text,Text,IntWritable>{private static final int MISSING = 9999;@Overridepublic void map(LongWritable key,Text value,Context context) throws IOException,InterruptedException{String line = value.toString();String year = line.substring(15,19);int airTemperature;if(line.charAt(87) =='+'){airTemperature = Integer.parseInt(line.substring(88,92));else{airTemperature = Integer.parseInt(line.subtring(87,92));
}String quality = line.substring(92,93);if(airTemperature != MISSING &&quality.matches("[01459]")){context.write(new Text(year),new IntWritetable(airTemperature));
}}}    }

这个Mapper类是一个泛型类型，他有四个行参类型，分别指定：map函数的输入键，输入值，输出键和输出值的类型。就现在的例子来说，输入键是一个长整数偏移量，输入值是一行文本，输出键是年份，输出值是气温(整数)。Hadoop本身提供了一套可优化网络序列化传输的基本类型，而不直接使用Java内嵌的类型。这些类型都在org.apache.hadoop.io包中。这里使用LongWritable类型(相当于Java的Long类型)、Text类型(相当于Java的String类型)和IntWritable类型(相当于Java的Integer类型)。

map()方法的输入时一个键和一个值。我们首先将包含有一行输入的Text值转换成Java的String类型，之后使用substring()方法提取我们感兴趣的列。

map()方法还提供了Context实例用于输出内容的写入。在这种情况下，我们将年份按Text对象进行读/写(因为我们把年份当作键)，将气温值封装在IntWritable类型中。只有气温数据不缺并且对应质量代码显示为正确的气温读数时，这些数据才会被写入输出记录中。

以类似方法用Reducer来定义reduce函数，如范例2-4所示。

范例2-4.查找最高气温的Reducer类

import  java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apche.hadoop.io.Text;
import org.apche.hadoop.mapreduce.Reducer;public class MaxTemperatureReducer extends Reducer<Text,IntWritable,Text,IntWritable>{@Overidepublic void reduce(Text key,Interable<IntWritable> values,Context context){int maxValue = Integer.MIN_VALUE;for(IntWritable value:values){maxValue = Max.max(maxValue,value.get());
}
context.write(key,new IntWritable(maxValue));}}

同样，reduce函数也有四个形式参数类型用于指定输入和输出类型。reduce函数的输入类型必须匹配map函数的输出类型:即Text类型和IntWritable类型。在这种情况下，reduce函数的输出类型也必须是Text和IntWritable类型，分别输出年份及其最高气温。这个最高气温是通过循环比较每个气温与当前所知最高气温所得到的。

第三部分代码负责运行MapReduce作业(请参见范例2-5)

范例2-5 这个应用程序在气象数据集中找出最高气温

import java.oo.IOException;
import org.apache.hadoop.fs.Path;
import org.apahce.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.input.FileOutputFormat;
import org.apache.hadoop.mapreduce.input.FileOutputFormat;public class MaxTemperature{public static void main(String[] args) throws Exception{if(args.length !=2){System.err.printlin("Usage:MaxTemperature<input path> <output path>");
System.exit(-1);}Job job  = new Job();
job.serJarByClass(MaxTemperature.class);
job.setName("Max temperature");FileInputFormat.addInputPath(job,new Path(args[0]));
FileOutputFormat.setOutputPath(job,new Path([args[1]));job.setMapperClass(MaxTemperatureMapper,class);
job.setReducerClass(MaxTemperatureReducer.class);job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);System.exit(job.waitForCompletion(true)?0:1);}}

Job对象可以指定作业执行规范。我们可以用它来控制整个作业的运行。我们在Hadoop集群上运行这个作业时，要把代码打包成一个Jar文件(Haoop在集群上发布在合格文件)。

不必明确指定JAR文件的名称，在Job对象的setJarByClass()方法中传递一个类即可，Hadoop利用这个类来查找包含它的JAR文件，进而找到相关的Jar文件。

构造Job对象之后，需要指定输入和输出数据的路径。调用FileInputFormat类的静态方法addInputPath()来定义输入数据的路径，这个路径可以是单个的文件、一个目录(此时，将目录下所有文件当做输入)或符合特定文件模式的一系列文件。由函数名可知，可以多次调用addInputPath()来实现多路径的输入。

调用FileOutputFormat类中的静态方法setOutputPath()来指定输出路径(只能有一个输出路径)。这个方法指定的是reduce函数输出文件的写入目录。在运行作业前该目录是不应该存在的，否则Hadoop会报错并拒绝运行作业。这种预防措施的目的是放置数据丢失(长时间运行的作业如果结果被意外覆盖，肯定是非常恼人的)。

接着，调用setMapperClas()和setReducerClass()指定map类型和reduce类型。
setOutputKeyClass()和setOutputValueClass()控制map和reduce函数的输出类型，正如本例所示，这两个输出类型一般都是相同的。如果不同，则通过setMapOutputKeyClass()和setMapOutputValueClass()来设置mao函数的输出类型。

输入的类型通过InputFormat类来控制，我们的例子中没有设置，因为使用的是默认的TextInputFormat(文本输入格式)。

在设置定义map和reduce函数的类之后，可以开始运行作业。Job中的waitForCompletion()方法提交作业并等待执行完成。该方法中的布尔参数是个详细标识，所以作业会把进度写到控制台。

waitForCompletion()方法返回一个布尔值，表示执行的成(true)败(false)。

转载于:https://www.cnblogs.com/jingblogs/p/5519349.html

Java 实现MapReduce函数相关推荐

java mapreduce程序_简单的java Hadoop MapReduce程序(计算平均成绩)从打包到提交及运行...
[TOC] 简单的java Hadoop MapReduce程序(计算平均成绩)从打包到提交及运行程序源码 import java.io.IOException; import java.util. ...
java main 方法不能执行,AndroidStudio无法执行Java的main函数
AndroidStudio无法执行Java的main函数 AndroidStudio无法执行Java的main函数 FAILURE: Build failed with an exception. * ...
java虚成员函数_Java常见知识点汇总（④）——虚函数、抽象函数、抽象类、接口...
一. Java虚函数虚函数的存在是为了多态. 它虚就虚在所谓"推迟联编"或者"动态联编"上,一个类函数的调用并不是在编译时刻被确定的,而是在运行时刻被确定的. ...
java string()函数_转载java String.split()函数的用法详解
转载java String.split()函数的用法详解如果您发现本文排版有问题,可以先点击下面的链接切换至老版进行查看!!!在java.lang包中有String.split()方法的原型是: p ...
C#、C++、JAVA中虚函数和抽象函数的概念对比
这两天恶补了下C#的相关内容,虚函数的概念又挖掘了下,总结如下: 1.C++中函数被定义为虚函数的方法是在函数名前加virtual,虚函数可以有自己的具体内容,也可以不含有函数实现内容,称为纯虚函数. ...
不使用java内置函数，将String字符串转换为int类型
package com.test;public class AtoiTest {public static void main(String[] args) throws Exception {Str ...
java中main函数解析
作者:xwdreamer 出处:http://www.cnblogs.com/xwdreamer 欢迎任何形式的转载,但请务必注明出处. 从写java至今,写的最多的可能就是主函数 public st ...
Java中的函数传递
转载自 Java中的函数传递在C和C++中,函数的传递可以通过函数指针来实现.在C#中,函数传递可以通过委托.Action.Func来实现.Java中没有函数指针.没有委托,那函数要如何传递呢? ...
java中的de是什么_【转】java中main函数解析
源地址:http://www.cnblogs.com/xwdreamer/archive/2012/04/09/2438845.html 从写java至今,写的最多的可能就是主函数 public st ...

Java 实现MapReduce函数

Java 实现MapReduce函数相关推荐

最新文章

热门文章