Hadoop学习笔记-MapperReduce简介

lanying56123

Hadoop学习笔记-MapperReduce简介
　　

　　前言：
　　本文将对Hadoop里面的一个核心木块—MapperReduce进行一定的了解。主要是从map函数和reduce函数以及数据是如果在这里两个函数之间流转的，以及分布式的处理过程。介绍可能比较简单，毕竟我也是菜鸟，希望有不对之处还望指出。

一个简单字母统计的例子

本文并没有采用《Hadoop权威指南》里面统计最高气温的例子，而是用了一个比较简单的统计字母个数的例子。本实例包括三个class文件，文件组织组织结构如下：

　　FindSameAlphaMapper就是一个map操作，相应的FindSameAlphaReduce是一个reduce操作，而map和reduce则是交给一个Job来调用，则有JobRun来调用。
　　代码内容如下：
　　FindSameAlphaMapper.java

package com.jdream314.mapper;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
/**
*
* <pre>
* Mapper
* </pre>
* @author xxx  
* @version 1.00.00
* <pre>
* 修改记录
*    修改后版本:     修改人：  修改日期:     修改内容:
* </pre>
*/
public class FindSameAlphaMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
/* (non-Javadoc)
* @see org.apache.hadoop.mapreduce.Mapper#map(java.lang.Object, java.lang.Object, org.apache.hadoop.mapreduce.Mapper.Context)
*/
private Text word = new Text();
private final static IntWritable one = new IntWritable(1);
@Override
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String text = value.toString();
for(char al:text.toCharArray()){
context.write(new Text(String.valueOf(al)),new IntWritable(1));
}
}
}

　　FindSameAlphaReduce.java

package com.jdream314.reduce;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
/**
*
* <pre>
* reduce
* </pre>
* @author XXX
* @version 1.00.00
* <pre>
* 修改记录
*    修改后版本:     修改人：  修改日期:     修改内容:
* </pre>
*/
public class FindSameAlphaReduce extends Reducer<Text, IntWritable, Text, IntWritable> {

/* (non-Javadoc)
* @see org.apache.hadoop.mapreduce.Reducer#reduce(java.lang.Object, java.lang.Iterable, org.apache.hadoop.mapreduce.Reducer.Context)
*/
@Override
public void reduce(Text key, Iterable<IntWritable> values,Context context)
throws IOException, InterruptedException {
// TODO Auto-generated method stub
int sum=0;
for(IntWritable value:values){
sum+=value.get();
}
context.write(key, new IntWritable(sum));
}
}


　　JobRun.java

package com.jdream314.run;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
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 org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.jdream314.mapper.FindSameAlphaMapper;
import com.jdream314.reduce.FindSameAlphaReduce;

/**
*
* <pre>
* job
* </pre>
* @author XXX
* @version 1.00.00
* <pre>
* 修改记录
*    修改后版本:    修改人：  修改日期:     修改内容:
* </pre>
*/
publicclass JobRunextends Configured implements Tool {


@Override
publicint run(String[] arg0) throws Exception {
Job job  = Job.getInstance(getConf());
job.setJarByClass(JobRun.class);
job.setJobName("worldcount");

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

job.setMapperClass(FindSameAlphaMapper.class);
job.setCombinerClass(FindSameAlphaReduce.class);
job.setReducerClass(FindSameAlphaReduce.class);


job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);

FileInputFormat.setInputPaths(job, new Path("D:\\HADOOP\\input"));
FileOutputFormat.setOutputPath(job, new Path("D:\\HADOOP\\output"));

boolean success = job.waitForCompletion(true);
return success?0:1;
}

publicstaticvoid main(String[] args) throws Exception{
int ret = ToolRunner.run(new JobRun(), args);
System.exit(ret);
}

}
　　在D盘hadoop的文件结构是这样的


　　这里面我是有两个文件file01和file02。Hadoop会处理input文件下的所有文件内容，然后输出到output的目录中，注意，不要创建output的目录，只要指明就可以。执行完之后就会再output中出现如下：


　　而part-r-00000就是处理的结果。该文件不能通过记事本打开直接查看，需要通过cygwin的cat查看，如下：


　　前面子字母，后面是出现的次数

MapperReduce介绍
　　可以看出整个代码是比较简单的，在Mapper里面主要是对给出的原始数据进行处理，此处是分开每个字母，然后以字母为key，以个数为value。当然，每个这样一个键值对的value初始值都是为1，在mapper中不对字母进行统计，只是进行一个简单的处理，总的来说：mapper只是将一些非结构化或者是半结构化的数据进行了一定的结构化处理。那么在reduce中进行的则是对mapper处理后的数据进行分析。
　　从上面描述中可以大约的知道mapper和reduce的基本执行过程，但是mapper和reduce的执行是有一个Job来调度的。本实例只是一个简单的实例，并没有把hadoop的分布式以及多个节点并行处理的能力体现出来。那么hadoop是如果多节点并行处理mapper和reduce的呢？下面给出相关的基本描述。
　　在hadoop中，整个一个执行过程就是一个Job，那么原始数据交给job，然后分给mapper，再转交给reduce。整个数据的流转是这样一个过程的。那么hadoop是在哪里进行多节点集成的呢？
　　首先原始数据是一个大数据，可能来自不同的节点或者多个磁盘，首先mapper处理的数据如果是在当前节点肯定比在其他节点的速度要快，因为如果不在同一个节点，那么就会受到网络带宽的限制。这就是在《Hadoop权威指南》里面指出的数据局部性优化。Hadoop拿到原始数据做的第一件事就是分块，将大数据分成多个块，而每个块的大小最理想情况是默认的64M，那么每个块就会启动一个mapper任务，那么这里就体现了hadoop在mapper端的并行操作。先是将原始数据进行分块，然后通过块的个数来并行执行mapper任务。这样相比一个线程在处理整个原始数据的速度肯定快。注意：在mapper处理后的数据不会直接存储在hdfs上面，因为mapper处理的结果并不是最终的数据，他只是一个中间数据，那么这些数据就存储在本地的临时文件中，当执行问mapper操作之后将进行删除。这里就是mapper的多线程执行的操作。
　　到这里，不禁让我想起了cuda里面的分块，然后每个块里面有多个线程。这样就可以将任务分解到不同块的不同线程上执行。而CUDA里面的分块并不是对数据直接分块，而是间接的分块，为什么是间接的呢？CUDA里面不想Hadoop一样有具体的mapper函数或者是reduce函数。CUDA里面只有一个每个线程都会执行的函数，并且函数会接收当前处理该线程是哪个块以及这个块的哪个线程这些信息，那么更具这些信息就可以确定这个线程是处理哪些数据，那么这里的哪些数据是我们程序员自己定的，就像Hadoop里面mapper里面的键值对是程序员自己定义一样。所以这里可以看出CUDA是间接的对数据分块，而是通过直接的线程分块来导致数据的间接分块。而hadoop是首先就对数据进行分块，然后交给mapper任务处理。这只是我个人的一个联想，和hadoop没有太大的关系，扯了一会蛋！
　　上面对mapper分块进行了简单的介绍，现在就reduce进行相关的介绍。上面给出过描述，reduce是处理mapper处理之后的数据。那么reduce的整体执行肯定是在所有的mapper任务执行完之后再执行的，不然可能会导致数据的不完整。所以可以想象，在mapper执行完之后，有一个收集所有mapper任务执行的结果操作。然后将收集好的完整数据转移给reduce进行再次处理。那么reduce是不是可以和mapper一样进行并行的处理呢？答案是肯定的。但是reduce的任务个数，本人暂时还没了解到，后面将会给出相关的解释。Mapper执行完之后会产生很多中间数据，那么hadoop收集好这些中间数据之后将会交给reduce进行处理。当面hadoop也可以对这些中间数据进行分块处理。对中间数据进行分块，并且每个键的所有值必须在同一个块中，这个是hadoop在此处分块的必须条件，不然会导致处理结果的错误。那么reduce处理完毕之后将会统一提交给hdfs来进行存储。下面给出几个先关的数据走向图，看完之后应该会更加的清晰。



　　从上图可以看到，mapper处理完毕之后需要将数据提交给reduce任务。当执行mapper任务的节点和执行reduce的节点不是同一个的时候，那么这里的就会考虑网络带宽的问题，这个时候网络带宽会影响hadoop处理的效率。那么如何减小这些影响呢？那就是减少数据的传输，就是减少从mapper传递到reduce任务的数据量。这里就引出了combiner函数。combiner的执行次数不会影响reduce的结果。我们在这里引用一下《Hadoop权威指南》中的一个例子来说明combiner的功能。在《Hadoop权威指南》中有一个取得最高气温的例子，那么这里假设有两个mapper来处理这个需求分别得到下面两个数据：
　　第一个mapper:


　　第二个mapper：



那么传入到reduce的值应该是：


所以就可以得出最高气温是25。但是这里我们是将五个气温值传递给reduce。假设这个时候网络带宽很小，就连这五个值传递都很慢（这种情况或许真的只有在天朝才有可能）。那么可以通过combiner来减少传递的值，或者甚至不传递。

　　上图中红线框住的就是combiner处理后的结果，之前需要传递五个数字，那么此时只要传递两个。如果我们在进行一次combiner那么就得到了我们想要的结果。所以combiner的调用次数就看实际情况的需求。于是就出现了下图：

　　到此，对MapperReduce的介绍已经全部完毕了，后续会对Hadoop的另一个核心模块HDFS进行简单的介绍。

本问中的实例代码在我的资源中有下载，不需要积分！