1.yarn源码分析（四）AppMaster启动
2.如何分布式运行mapreduce程序
3.MapReduce源码解析之Mapper
4.如何在MaxCompute上运行HadoopMR作业
5.Mapreduce 在通过reduce计算value之后怎么统计计算次数？
6.MapReduce源码解析之InputFormat

yarn源码分析（四）AppMaster启动

       在容器分配完成之后，启动容器的代码主要在ContainerImpl.java中进行。通过状态机转换，container从NEW状态向其他状态转移时，会调用RequestResourceTransition对象。RequestResourceTransition负责将所需的百宝箱 源码资源进行本地化，或者避免资源本地化。若需本地化，还需过渡到LOCALIZING状态。为简化理解，此处仅关注是否进行资源本地化的情况。

       为了将LAUNCH_CONTAINER事件加入事件处理队列，调用了sendLaunchEvent方法。荆门直播源码该事件由ContainersLauncher负责处理。ContainersLauncher的handle方法中，使用一个ExecutorService（线程池）容器Launcher。ContainerLaunch实现了Callable接口，其call方法生成并执行launch_container脚本。以MapReduce框架为例，该脚本在hadoop.tmp.dir/application name/container name目录下生成，其主要作用是启动MRAppMaster进程，即MapReduce的ApplicationMaster。

如何分布式运行mapreduce程序

       ä¸€ã€ 首先要知道此前提 转载

       ã€€ã€€è‹¥åœ¨windows的Eclipse工程中直接启动mapreduc程序，需要先把hadoop集群的配置目录下的xml都拷贝到src目录下，让程序自动读取集群的地址后去进行分布式运行(您也可以自己写java代码去设置job的configuration属性)。

       ã€€ã€€è‹¥ä¸æ‹·è´ï¼Œå·¥ç¨‹ä¸­bin目录没有完整的xml配置文件，则windows执行的mapreduce程序全部通过本机的jvm执行，作业名也是带有“local"字眼的作业，如 job_local_。 这不是真正的分布式运行mapreduce程序。

       ã€€ã€€ä¼°è®¡å¾—研究org.apache.hadoop.conf.Configuration的源码，反正xml配置文件会影响执行mapreduce使用的文件系统是本机的windows文件系统还是远程的hdfs系统; 还有影响执行mapreduce的mapper和reducer的是本机的jvm还是集群里面机器的jvm

       ã€€ã€€äºŒã€ 本文的结论

       ã€€ã€€ç¬¬ä¸€ç‚¹å°±æ˜¯ï¼š windows上执行mapreduce，必须打jar包到所有slave节点才能正确分布式运行mapreduce程序。（有个需求是要windows上触发一个mapreduce分布式运行）

       ã€€ã€€ç¬¬äºŒç‚¹å°±æ˜¯ï¼š Linux上，只需拷贝jar文件到集群master上,执行命令hadoop jarPackage.jar MainClassName即可分布式运行mapreduce程序。

       ã€€ã€€ç¬¬ä¸‰ç‚¹å°±æ˜¯ï¼š 推荐使用附一，实现了自动打jar包并上传，分布式执行的mapreduce程序。

       ã€€ã€€é™„一、 推荐使用此方法：实现了自动打jar包并上传，分布式执行的mapreduce程序：

       ã€€ã€€è¯·å…ˆå‚考博文五篇：

       ã€€ã€€Hadoop作业提交分析（一）~~（五）

       ã€€ã€€å¼•ç”¨åšæ–‡çš„附件中EJob.java到工程中，然后main中添加如下方法和代码。

       ã€€ã€€public static File createPack() throws IOException {

       ã€€ã€€File jarFile = EJob.createTempJar("bin");

       ã€€ã€€ClassLoader classLoader = EJob.getClassLoader();

       ã€€ã€€Thread.currentThread().setContextClassLoader(classLoader);

       ã€€ã€€return jarFile;

       ã€€ã€€}

       ã€€ã€€åœ¨ä½œä¸šå¯åŠ¨ä»£ç ä¸­ä½¿ç”¨æ‰“包：

       ã€€ã€€Job job = Job.getInstance(conf, "testAnaAction");

       ã€€ã€€æ·»åŠ ï¼š

       ã€€ã€€String jarPath = createPack().getPath();

       ã€€ã€€job.setJar(jarPath);

       ã€€ã€€å³å¯å®žçŽ°ç›´æŽ¥run as java application 在windows跑分布式的mapreduce程序，不用手工上传jar文件。

       ã€€ã€€é™„二、得出结论的测试过程

       ã€€ã€€ï¼ˆæœªæœ‰ç©ºçœ‹ä¹¦ï¼Œåªèƒ½é€šè¿‡æ„šç¬¨çš„测试方法得出结论了）

       ã€€ã€€ä¸€. 直接通过windows上Eclipse右击main程序的java文件，然后"run as application"或选择hadoop插件"run on hadoop"来触发执行MapReduce程序的测试。

       ã€€ã€€1，如果不打jar包到进集群任意linux机器上，它报错如下：

       ã€€ã€€[work] -- ::, - org.apache.hadoop.mapreduce.Job - [main] INFO org.apache.hadoop.mapreduce.Job - map 0% reduce 0%

       ã€€ã€€[work] -- ::, - org.apache.hadoop.mapreduce.Job - [main] INFO org.apache.hadoop.mapreduce.Job - Task Id : attempt___m__0, Status : FAILED

       ã€€ã€€Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found

       ã€€ã€€at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:)

       ã€€ã€€at org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:)

       ã€€ã€€at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:)

       ã€€ã€€at org.apache.hadoop.mapred.MapTask.run(MapTask.java:)

       ã€€ã€€at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:)

       ã€€ã€€at java.security.AccessController.doPrivileged(Native Method)

       ã€€ã€€at javax.security.auth.Subject.doAs(Subject.java:)

       ã€€ã€€at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:)

       ã€€ã€€at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:)

       ã€€ã€€Caused by: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found

       ã€€ã€€at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:)

       ã€€ã€€at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:)

       ã€€ã€€... 8 more

       ã€€ã€€# Error:后重复三次

       ã€€ã€€-- ::, - org.apache.hadoop.mapreduce.Job - [main] INFO org.apache.hadoop.mapreduce.Job - map % reduce %

       ã€€ã€€çŽ°è±¡å°±æ˜¯ï¼šæŠ¥é”™ï¼Œæ— è¿›åº¦ï¼Œæ— è¿è¡Œç»“果。

       ã€€ã€€

       ã€€ã€€2，拷贝jar包到“只是”集群master的$HADOOP_HOME/share/hadoop/mapreduce/目录上，直接通过windows的eclipse "run as application"和通过hadoop插件"run on hadoop"来触发执行，它报错同上。

       ã€€ã€€çŽ°è±¡å°±æ˜¯ï¼šæŠ¥é”™ï¼Œæ— è¿›åº¦ï¼Œæ— è¿è¡Œç»“果。

       ã€€ã€€3，拷贝jar包到集群某些slave的$HADOOP_HOME/share/hadoop/mapreduce/目录上，直接通过windows的eclipse "run as application"和通过hadoop插件"run on hadoop"来触发执行

       ã€€ã€€å’ŒæŠ¥é”™ï¼š

       ã€€ã€€Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found

       ã€€ã€€at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:)

       ã€€ã€€at org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:)

       ã€€ã€€å’ŒæŠ¥é”™ï¼š

       ã€€ã€€Error: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountReducer not found

       ã€€ã€€

       ã€€ã€€çŽ°è±¡å°±æ˜¯ï¼šæœ‰æŠ¥é”™ï¼Œä½†ä»ç„¶æœ‰è¿›åº¦ï¼Œæœ‰è¿è¡Œç»“果。

MapReduce源码解析之Mapper

       MapReduce，大数据领域的标志性计算模型，由Google公司研发，插件源码bug其核心概念"Map"与"Reduce"简明易懂却威力巨大，打开了大数据时代的大门。对于许多大数据工作者来说，MapReduce是基础技能之一，而源码解析更是深入理解与实践的必要途径。

       MapReduce由两部分组成：Map与Reduce。Map阶段通过映射函数将一组键值对转换成另一组键值对，而Reduce阶段则负责合并这些新的键值对。这种并行计算模型极大地提高了大数据处理的效率。

       本文将聚焦于Map阶段的核心实现——Mapper。通过解析Mapper类及其子类的源码，我们可以更深入地理解MapReduce的电脑时钟源码工作机制，并在易观千帆等技术数据处理中发挥更大的效能。

       Mapper类内部包含四个关键方法与一个抽象类：

       setup()：主要为map()方法做准备，例如加载配置文件、传递参数。

       cleanup()：用于清理资源，如关闭文件、处理Key-Value。

       map()：程序的逻辑核心，对输入的文本进行处理（如分割、过滤），以键值对的形式写入context。

       run()：驱动Mapper执行的新源码屋主方法，按照预设顺序执行setup()、map()、cleanup()。

       Context抽象类扮演着重要角色，用于跟踪任务状态和数据存储，如在setup()中读取配置信息，并作为Key-Value载体。

       下面是几个Mapper子类的详细解析：

       InverseMapper：将键值对反转，适用于不同需求的统计分析。

       TokenCounterMapper：使用StringTokenizer对文本进行分割，计算特定token的数量，适用于词频统计等。

       RegexMapper：对文本进行正则化处理，适用于特定格式文本的统计。

       MultithreadedMapper：利用多线程执行Mapper任务，提高CPU利用率，适用于并发处理。

       本文对MapReduce中Mapper及其子类的源码进行了详尽解析，旨在帮助开发者更深入地理解MapReduce的实现机制。后续将探讨更多关键类源码，以期为大数据处理提供更深入的洞察与实践指导。

如何在MaxCompute上运行HadoopMR作业

       MaxCompute（原ODPS）有一套自己的MapReduce编程模型和接口，简单说来，这套接口的输入输出都是MaxCompute中的Table，处理的数据是以Record为组织形式的，它可以很好地描述Table中的数据处理过程，然而与社区的Hadoop相比，编程接口差异较大。Hadoop用户如果要将原来的Hadoop MR作业迁移到MaxCompute的MR执行，需要重写MR的代码，使用MaxCompute的接口进行编译和调试，运行正常后再打成一个Jar包才能放到MaxCompute的平台来运行。这个过程十分繁琐，需要耗费很多的开发和测试人力。如果能够完全不改或者少量地修改原来的Hadoop MR代码就能在MaxCompute平台上跑起来，将是一个比较理想的方式。

       çŽ°åœ¨MaxCompute平台提供了一个HadoopMR到MaxCompute MR的适配工具，已经在一定程度上实现了Hadoop MR作业的二进制级别的兼容，即用户可以在不改代码的情况下通过指定一些配置，就能将原来在Hadoop上运行的MR jar包拿过来直接跑在MaxCompute上。目前该插件处于测试阶段，暂时还不能支持用户自定义comparator和自定义key类型，下面将以WordCount程序为例，介绍一下这个插件的基本使用方式。

       ä½¿ç”¨è¯¥æ’件在MaxCompute平台跑一个HadoopMR作业的基本步骤如下：

       1. 下载HadoopMR的插件

       ä¸‹è½½æ’件，包名为hadoop2openmr-1.0.jar，注意，这个jar里面已经包含hadoop-2.7.2版本的相关依赖，在作业的jar包中请不要携带hadoop的依赖，避免版本冲突。

       2. 准备好HadoopMR的程序jar包

       ç¼–译导出WordCount的jar包：wordcount_test.jar ，wordcount程序的源码如下:

       package com.aliyun.odps.mapred.example.hadoop;

       import org.apache.hadoop.conf.Configuration;

       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.Mapper;

       import org.apache.hadoop.mapreduce.Reducer;

       import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

       import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

       import java.io.IOException;

       import java.util.StringTokenizer;

       public class WordCount {

       public static class TokenizerMapper

       extends Mapper<Object, Text, Text, IntWritable>{

       private final static IntWritable one = new IntWritable(1);

       private Text word = new Text();

       public void map(Object key, Text value, Context context

       ) throws IOException, InterruptedException {

       StringTokenizer itr = new StringTokenizer(value.toString());

       while (itr.hasMoreTokens()) {

       word.set(itr.nextToken());

       context.write(word, one);

       }

       }

       }

       public static class IntSumReducer

       extends Reducer<Text,IntWritable,Text,IntWritable> {

       private IntWritable result = new IntWritable();

       public void reduce(Text key, Iterable<IntWritable> values,

       Context context

       ) throws IOException, InterruptedException {

       int sum = 0;

       for (IntWritable val : values) {

       sum += val.get();

       }

       result.set(sum);

       context.write(key, result);

       }

       }

       public static void main(String[] args) throws Exception {

       Configuration conf = new Configuration();

       Job job = Job.getInstance(conf, "word count");

       job.setJarByClass(WordCount.class);

       job.setMapperClass(TokenizerMapper.class);

       job.setCombinerClass(IntSumReducer.class);

       job.setReducerClass(IntSumReducer.class);

       job.setOutputKeyClass(Text.class);

       job.setOutputValueClass(IntWritable.class);

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

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

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

       }

       }

       3. 测试数据准备

       åˆ›å»ºè¾“入表和输出表

       create table if not exists wc_in(line string);

       create table if not exists wc_out(key string, cnt bigint);

       é€šè¿‡tunnel将数据导入输入表中

       å¾…导入文本文件data.txt的数据内容如下：

       hello maxcompute

       hello mapreduce

       ä¾‹å¦‚可以通过如下命令将data.txt的数据导入wc_in中，

       tunnel upload data.txt wc_in;

       4. 准备好表与hdfs文件路径的映射关系配置

       é…ç½®æ–‡ä»¶å‘½åä¸ºï¼šwordcount-table-res.conf

       {

       "file:/foo": {

       "resolver": {

       "resolver": "c.TextFileResolver",

       "properties": {

       "text.resolver.columns.combine.enable": "true",

       "text.resolver.seperator": "\t"

       }

       },

       "tableInfos": [

       {

       "tblName": "wc_in",

       "partSpec": { },

       "label": "__default__"

       }

       ],

       "matchMode": "exact"

       },

       "file:/bar": {

       "resolver": {

       "resolver": "openmr.resolver.BinaryFileResolver",

       "properties": {

       "binary.resolver.input.key.class" : "org.apache.hadoop.io.Text",

       "binary.resolver.input.value.class" : "org.apache.hadoop.io.LongWritable"

       }

       },

       "tableInfos": [

       {

       "tblName": "wc_out",

       "partSpec": { },

       "label": "__default__"

       }

       ],

       "matchMode": "fuzzy"

       }

       }

Mapreduce 在通过reduce计算value之后怎么统计计算次数？

       简单，不知道你看没看过Wordcount源码，其中的统计出现次数是传入一个1，通过reduce相加计算得出次数。我可以通过Map传入value时拼接一个1，在reduce中通过拆分字符串得到你要的原valeu和传入的1 ，分别去计算后再拼入输出就可以得到了

MapReduce源码解析之InputFormat

       导读

       深入探讨MapReduce框架的核心组件——InputFormat。此组件在处理多样化数据类型时，扮演着数据格式化和分片的角色。通过设置job.setInputFormatClass(TextInputFormat.class)等操作，程序能正确处理不同文件类型。InputFormat类作为抽象基础，定义了文件切分逻辑和RecordReader接口，用于读取分片数据。本节将解析InputFormat、InputSplit、RecordReader的结构与实现，以及如何在Map任务中应用此框架。

       类图与源码解析

       InputFormat类提供了两个关键抽象方法：getSplits()和createRecordReader()。getSplits()负责规划文件切分策略，定义逻辑上的分片，而RecordReader则从这些分片中读取数据。

       InputSplit类承载了切分逻辑，表示了给定Mapper处理的逻辑数据块，包含所有K-V对的集合。

       RecordReader类实现了数据读取流程，其子类如LineRecordReader，提供行数据读取功能，将输入流中的数据按行拆分，赋值为Key和Value。

       具体实现与操作流程

       在getSplits()方法中，FileInputFormat类负责将输入文件按照指定策略切分成多个InputSplit。

       TextInputFormat类的createRecordReader()方法创建了LineRecordReader实例，用于读取文件中的每一行数据，形成K-V对。

       Mapper任务执行时，通过调用RecordReader的nextKeyValue()方法，读取文件的每一行，完成数据处理。

       在Map任务的run()方法中，MapContextImp类实例化了一个RecordReader，用于实现数据的迭代和处理。

       总结

       本文详细阐述了MapReduce框架中InputFormat的实现原理及其相关组件，包括类图、源码解析、具体实现与操作流程。后续文章将继续探讨MapReduce框架的其他关键组件源码解析，为开发者提供深入理解MapReduce的构建和优化方法。