MapReduce算法并行处理单词种类的数量

蔚落 2022-09-27 06:21 62阅读 0赞

# 一、MapReduce简介 #

## 1.1 MapReduce编程模型 ##

MapReduce采用”分而治之”的思想，把对大规模数据集的操作，分发给一个主节点管理下的各个分节点共同完成，然后通过整合各个节点的中间结果，得到最终结果。简单地说，MapReduce就是”任务的分解与结果的汇总”。

在Hadoop中，用于执行MapReduce任务的机器角色有两个：

*  JobTracker用于调度工作的，一个Hadoop集群中只有一个JobTracker，位于master。
 *  TaskTracker用于执行工作，位于各slave上。

在分布式计算中，MapReduce框架负责处理了并行编程中分布式存储、工作调度、负载均衡、容错均衡、容错处理以及网络通信等复杂问题，把处理过程高度抽象为两个函数：map和reduce，map负责把任务分解成多个任务，reduce负责把分解后多任务处理的结果汇总起来。

需要注意的是，用MapReduce来处理的数据集（或任务）必须具备这样的特点：待处理的数据集可以分解成许多小的数据集，而且每一个小数据集都可以完全并行地进行处理。

## 1.2 MapReduce工作过程 ##

对于一个MR任务，它的输入、输出以及中间结果都是`<key, value>`键值对：

*  Map：`<k1, v1>` ——> `list(<k2, v2>)`
 *  Reduce：`<k2, list(v2)>` ——> `list(<k3, v3>)`

MR程序的执行过程主要分为三步：Map阶段、Shuffle阶段、Reduce阶段，如下图：

![20160811132825039][]

1.  **Map阶段**
    
     *  分片（Split）：map阶段的输入通常是HDFS上文件，在运行Mapper前，FileInputFormat会将输入文件分割成多个split ——1个split至少包含1个HDFS的Block（默认为64M）；然后每一个分片运行一个map进行处理。
     *  执行（Map）：对输入分片中的每个键值对调用`map()`函数进行运算，然后输出一个结果键值对。
        
         *  Partitioner：对`map()`的输出进行partition，即根据key或value及reduce的数量来决定当前的这对键值对最终应该交由哪个reduce处理。默认是对key哈希后再以reduce task数量取模，默认的取模方式只是为了避免数据倾斜。然后该key/value对以及partitionIdx的结果都会被写入环形缓冲区。
     *  溢写（Spill）：map输出写在内存中的环形缓冲区，默认当缓冲区满80%，启动溢写线程，将缓冲的数据写出到磁盘。
        
         *  Sort：在溢写到磁盘之前，使用快排对缓冲区数据按照partitionIdx, key排序。（每个partitionIdx表示一个分区，一个分区对应一个reduce）
         *  Combiner：如果设置了Combiner，那么在Sort之后，还会对具有相同key的键值对进行合并，减少溢写到磁盘的数据量。
     *  合并（Merge）：溢写可能会生成多个文件，这时需要将多个文件合并成一个文件。合并的过程中会不断地进行 sort & combine 操作，最后合并成了一个已分区且已排序的文件。
2.  **Shuffle阶段**：广义上Shuffle阶段横跨Map端和Reduce端，在Map端包括Spill过程，在Reduce端包括copy和merge/sort过程。通常认为Shuffle阶段就是将map的输出作为reduce的输入的过程
    
     *  Copy过程：Reduce端启动一些copy线程，通过HTTP方式将map端输出文件中属于自己的部分拉取到本地。Reduce会从多个map端拉取数据，并且每个map的数据都是有序的。
     *  Merge过程：Copy过来的数据会先放入内存缓冲区中，这里的缓冲区比较大；当缓冲区数据量达到一定阈值时，将数据溢写到磁盘（与map端类似，溢写过程会执行 sort & combine）。如果生成了多个溢写文件，它们会被merge成一个**有序的最终文件**。这个过程也会不停地执行 sort & combine 操作。
3.  **Reduce阶段**：Shuffle阶段最终生成了一个有序的文件作为Reduce的输入，对于该文件中的每一个键值对调用`reduce()`方法，并将结果写到HDFS。

二.一个用MapReduce算法实现的单词处理程序

MyMapper.java(Map部分)

package haige;
    
    import java.io.IOException;
    import java.util.StringTokenizer;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    public class MyMapper extends Mapper<Object, Text, Text, IntWritable> {
      private final static IntWritable one=new IntWritable();
      private Text word=new Text();
      @Override
      public void map(Object key, Text value,Context context) throws IOException,InterruptedException{
    	// TODO Auto-generated method stub
    	StringTokenizer itr=new StringTokenizer(value.toString());
    	while(itr.hasMoreTokens()){
    		word.set(itr.nextToken());
    		context.write(word, one);
    	}
    }
      
    }

MyReducer.java(Reduce部分)

package haige;
    
    import java.io.IOException;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    public class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    	private IntWritable result=new IntWritable();
    
    	@Override
    	protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
    		// TODO Auto-generated method stub
    		int sum=0;
    		for(IntWritable val : values){
    			sum+=val.get();
    		}
    		result.set(sum);
    		context.write(key,result);
    	}
    	
    }

WordCount.java(Main函数)

package haige;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.util.GenericOptionsParser;
    
    import com.sun.jersey.core.impl.provider.entity.XMLJAXBElementProvider.Text;
    
    public class WordCount {
    	public static void main(String[] args)throws Exception {
    		// TODO Auto-generated method stub
    		Configuration conf =new Configuration();
    		String [] otherArgs= new GenericOptionsParser(conf, args).getRemainingArgs();
    		if(otherArgs.length!=2){
    			System.err.println("Usage: wordcount <in><out>");
    			System.exit(2);
    		}
    		Job job=new Job(conf, "word count");  //设置环境参数
    		job.setJarByClass(WordCount.class);   
    		job.setMapperClass(MyMapper.class);    //添加MyMapper类
    		job.setReducerClass(MyReducer.class);  //添加MyReducer类
    		job.setOutputKeyClass(Text.class);     //设置输出类型(键值)
    		job.setOutputValueClass(IntWritable.class);  //设置输出类型(键值)
    		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));  //设置输入文件
    		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));  //设置输出文件
    		System.exit(job.waitForCompletion(true)?0:1);
    	}
    }

配置文件

core-site.xml

<?xml version="1.0" encoding="UTF-8"?>
    <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
    
    
    
    
    <configuration>
        <property>
            <name>hadoop.tmp.dir</name>
            <value>file:/usr/java/hadoop/tmp</value>
            <description>Abase for other temporary directories.</description>
        </property>
        <property>
            <name>fs.default.name</name>
            <value>hdfs://localhost:9000</value>
        </property>
    </configuration>

hdfs-site.xml

<?xml version="1.0" encoding="UTF-8"?>
    <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
    
    
    
    
    <configuration>
      <property>
            <name>dfs.replication</name>
            <value>1</value>
      </property>
      <property>
          <name>dfs.namenode.name.dir</name>
          <value>file:/usr/java/hadoop/tmp/dfs/name</value>
      </property>
      <property>
          <name>dfs.datanode.data.dir</name>
          <value>file:/usr/java/hadoop/tmp/dfs/data</value>
      </property>
      <property>   
           
        <name>dfs.permissions</name>   
        <value>false</value>   
      </property> 
    </configuration>

接着在DFS Location或输出的日志文件可以看到输出的output文件

[20160811132825039]: /images/20220708/910ea76c694f4c819224ed0e5e12772b.png