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方法

ReduceTask流程是怎样的

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);

ReduceTask流程是怎样的

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://scyanting.com/article/iejdhi.html