整套大数据学习资料(视频+笔记)百度网盘无门槛下载:http://www.edu360.cn/news/content?id=3377

2.3.2. JavaMapReduce

hadoop 花牛 11℃ 0评论

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

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

importjava.io.IOException;

 

importorg.apache.hadoop.io.IntWritable;

importorg.apache.hadoop.io.LongWritable;

importorg.apache.hadoop.io.Text;

importorg.apache.hadoop.mapreduce.Mapper;

 

publicclassMaxTempenatureMapperextendsMapper<LongWritable,Text,Text,IntWritable>{

 

privatestaticfinalintMISSING=9999;

 

@Override

publicvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{

Stringline=value.toString();

Stringyear=line.substring(15,19);

intairTemperature;

if(line.charAt(87)==’+’){//parselntdoesn’tlikeleadingplussigns

airTemperature=Integer.parseInt(line.substring(88,92));

}else{

airTemperature=Integer.parseInt(line.substring(87,92));

}

Stringquality=line.substring(92,93);

if(airTemperature!=MISSING&&quality.matches(“[01459]”)){

context.write(newText(year),newIntWritable(airTemperature));

}

}

}

 

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

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

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

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

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

importjava.io.IOException;

 

importorg.apache.hadoop.io.IntWritable;

importorg.apache.hadoop.io.Text;

importorg.apache.hadoop.mapreduce.Reducer;

 

publicclassMaxTemperatureReducerextendsReducer<Text,IntWritable,Text,IntWritable>{

@Override

publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)

throwsIOException,InterruptedException{

intmaxValue=Integer.MIN_VALUE;

for(IntWritablevalue:values){

maxValue=Math.max(maxValue,value.get());

}

context.write(key,newIntWritable(maxValue));

}

}

 

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

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

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

importorg.apache.hadoop.fs.Path;

importorg.apache.hadoop.io.IntWritable;

importorg.apache.hadoop.io.Text;

importorg.apache.hadoop.mapreduce.Job;

importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;

importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 

publicclassMaxTemperature{

publicstaticvoidmain(String[]args)throwsException{

if(args.length!=2){

System.err.println(“Usage:MaxTemperature〈inputpath>〈outputpath〉”);

System.exit(-1);

}

Jobjob=Job.getInstance();

job.setJarByClass(MaxTemperature.class);

job.setJobName(“Maxtemperature”);

FileInputFormat.addInputPath(job,newPath(args[0]));

FileOutputFormat.setOutputPath(job,newPath(args[1]));

job.setMapperClass(MaxTempenatureMapper.class);

job.setReducerClass(MaxTemperatureReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

System.exit(job.waitForCompletion(true)?0:1);

}

}

 

Job对象指定作业执行规范。我们可以用它来控制整个作业的运行。我们在Hadoop集群上运行这个作业时,要把代码打包成一个JAR文件(Hadoop在集群上发布这个文件)。不必明确指定JAR文件的名称,在Job对象的setJarByClass()方法中传递一个类即可,Hadoop利用这个类来査找包含它的JAR文件,进而找到相关的JAR文件。

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

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

接着,通过setMapperClass()和setReducerClass()指定map类型和reduce类型。

setOutputKeyClass()和setOutputValueClass()控制map和reduce函数的输出类型,正如本例所示,这两个输出类型一般都是相同的。如果不同,则通过setMapOutputKeyClass()和setMapOutputValueClass()来设置map函数的输出类型。

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

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

waitForCompletion()方法返回一个布尔值,表示执行的成(true)败(false),这个布尔值被转换成程序的退出代码0或者1。

转载请注明:全栈大数据 » 2.3.2. JavaMapReduce

喜欢 (0)or分享 (0)
发表我的评论
取消评论

表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址