ReduceTask流程是怎样的

这篇文章主要介绍“ReduceTask流程是怎样的”，在日常操作中，相信很多人在ReduceTask流程是怎样的问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”ReduceTask流程是怎样的”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

目前创新互联已为上1000+的企业提供了网站建设、域名、网络空间、成都网站托管、企业网站设计、清原网站维护等服务，公司将坚持客户导向、应用为本的策略，正道将秉承"和谐、参与、激情"的文化，与客户和合作伙伴齐心协力一起成长，共同发展。

ReduceTask流程源码解读

1、最终的文件就是 file.out 和 file.out.index ,等待reduce的拷贝.

2、在LocalJobRunner$Job中的run方法中: 		//LocalJobRunner类中555行
if (numReduceTasks > 0) {  			//判断reduceTask的个数
 	//创建Runnable对象: LocalJobRunner$Job$ReduceTaskRunnable
       	List reduceRunnables = getReduceTaskRunnables(
            jobId, mapOutputFiles);
        //创建线程池
	ExecutorService reduceService = createReduceExecutor();
	//将所有的LocalJobRunner$Job$ReduceTaskRunnable 提交到线程池执行.
        runTasks(reduceRunnables, reduceService, "reduce");
       }

3、进入runTasks(reduceRunnables, reduceService, "reduce");方法	//559行
for (Runnable r : runnables) {		 //循环每个Runnable,提交给线程池去执行.
    service.submit(r);
}

4、线程执行的时候，要运行LocalJobRunner$Job$ReduceTaskRunnable 中run方法

5、创建ReduceTask对象		//LocalJobRunner类~332行
ReduceTask reduce = new ReduceTask(systemJobFile.toString(),reduceId, taskId,mapIds.size(), 1);

6、执行ReduceTask中的run方法
//LocalJobRunner类 --> 347行reduce.run(localConf, Job.this); --> //进入run方法

7、调到ReduceTask的run方法内				//ReduceTask类~320行	
initialize(job, getJobID(), reporter, useNewApi);	//初始化~333行
sortPhase.complete(); 					//排序~382行
RawComparator comparator = job.getOutputValueGroupingComparator();	//387行 获取分组比较器

8、进入下列代码（390行）
runNewReducer(job, umbilical, reporter, rIter, comparator, keyClass, valueClass);
进入runNewReducer方法内		//ReduceTask~577行

--获取job的相关信息
 org.apache.hadoop.mapreduce.TaskAttemptContext taskContext =
      new org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl(job, getTaskID(), reporter);
 
--反射的操作创建reduce对象 ，例如: WordCountReducer
org.apache.hadoop.mapreduce.Reducer reducer =
      (org.apache.hadoop.mapreduce.Reducer)
        ReflectionUtils.newInstance(taskContext.getReducerClass(), job);
        
--创建RecordWriter对象
org.apache.hadoop.mapreduce.RecordWriter trackedRW = 
      new NewTrackingRecordWriter(this, taskContext);

9、向下走，定位到reducer.run(reducerContext);方法 --> 然后进入（Reducer的run方法）	//~628行
setup(context);
reduce(context.getCurrentKey(), context.getValues(), context); 
//执行到WordCountReducer中的reduce方法，是一个循环调用过程.
context.write(key,outv);		//数据写出源码流程如下：
①：reduceContext.write(key, value);
②：output.write(key, value);

//进入到ReduceTask的write方法		//557行
③：real.write(key,value); 		//real :TextOutputFormat$LineRecordWriter
进入到real.write()方法			//TextOutputFormat类~84行
writeObject(key);   //写出key
writeObject(value); //写出value

写出key的源码~简单看下：			//TextOutputFormat类~75行
private void writeObject(Object o) throws IOException {
	if (o instanceof Text) {
		Text to = (Text) o;
		out.write(to.getBytes(), 0, to.getLength());
	} else {
		out.write(o.toString().getBytes(StandardCharsets.UTF_8));
		//调用对象的toString方法，将返回的字符串转换成字节,通过流写出
	}
}

10、cleanup(context);   		//清除生相关的文件，生成分区文件

整体MR工作机制源码解读总结

源码总结说明:
1. 看源码目的:
       熟悉整个MR的流程，能够将我们讲解的知识点对应到源码中具体的位置.
       为面试做准备.

2. 在整个MR中 ,会有N个MapTask（按照切片数量决定个数）和 N个ReduceTask（自行设置个数)
	--在集群中的效果是多个MapTask并行运行, 并行数由集群的资源来决定. 
	--多个ReduceTask并行运行，并行数由集群的资源来决定. 一般来说，ReduceTask的数量比较少，基本上都
能够同时并行.

到此，关于“ReduceTask流程是怎样的”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注创新互联网站，小编会继续努力为大家带来更多实用的文章！

本文名称：ReduceTask流程是怎样的
文章地址：http://6mz.cn/article/iejdhi.html