CENTOS上的网络安全工具（十一）走向Hadoop（3） MapReduce示例解析

        当然，也可以自己构建xml文件放在同目录下，使用Configuration进行操作——然而，这些内容我们就不打算关注了。理解mapreduce只需要知道，Configuration记录了hadoop的配置信息，而mapreduce任务如何分解、如何分发等等是需要使用这些信息的，所以，每个job里面都需要携带并传递Configuration的句柄，以确保想知道配置的时候能够读得到。

        Configuration类实现了Iterable和Writable接口。Hadoop.io.Writable主要是2个方法：

                方法1：write(Java.io.DataOutput arg)

                方法2：readFields(Java.io.DataInput arg)

        Java.lang.Iterable主要是2个方法：

                方法1：forEach(Consumer action)

                方法2：spliterator()

       从此也可见Configuration的主要功能就是读写枚举配置信息。

       2.Configurable接口和Configured类

        Configured类用于具体实现Configurable接口

        上图来自org.apache.hadoop.conf-Configured_weixin_34148508的博客-CSDN博客 文章。

        Configurable接口主要维护一个私有属性：

        属性1：private Configuration conf

        并定义2个方法：

        方法1：setConf(Configuration conf)

        方法2：getConf(Configuration conf)

        可见，该接口主要确定继承自Configured类的类可维护Configuration配置环境，并提供对该配置环境的管理手段。

        3.Tool接口

public interface Tool extends Configurable {
    int run(String [] args) throws Exception;
    }
}

Public class Configured implements  Configurable{
    Private Configuration conf;
    Public Configured(Configuration conf){setConf(conf);}//构造方法
    Public void setConf(Configuration conf)
    {
        This.conf=conf;
    }
    Public getConf()
    {
        Return conf;
    }
}

           Tool接口继承自Configurable接口，新增了1个方法：

        方法1：run(String[] args)

        由于后面可以看到，ToolRunner会调用实现MapReduce算法的类的run方法，以启动一个MapReduce工作。所以，实现MapReduce算法的类需要实现Tool接口。

        在Tool类的注释中，给出了Tool.run()的典型实现方法：

public class MyApp extends Configured implements Tool {
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();//获取Hadoop环境
        JobConf Job = new JobConf(conf,MyApp.class);//构建MR环境
        Path in = new Path(args[1])
        Path out = new Path(args[2])
        //设置MapReduce任务
        Job.setJobName(“my-app”);
        Job.setInputPath(in);      //该方法并不存在，可能是伪代码
        Job.setOutputPath(out);     //该方法并不存在，可能是伪代码
        Job.setMapperClass(MyMapper.class)
        Job.setReducerClass(MyReducer.class)
        //提交MapReduce任务并等待结束
        RunningJob runningJob = JobClient.runJob(Job);
        if(runningJob.isSuccessful())return 0;
        else return 1;
    }
    public static void main(String[] args) throws Exception{
        //新建Hadoop环境，新建MR任务，并使用ToolRunner启动
        int res = ToolRunner.run(new Configuration(),new MyApp(),args);
        system.exit(res);
    }
}

        4.Grep类

        Grep类用于实现“Grep”MapReduce算法，所以需要实现Tool接口；且如前所述，Tool.run()中需要使用Hadoop的环境配置，故Grep也需要实现Configurable接口。所以，Grep的定义方式是：

        public class Grep extends Configurable implements Tool{}

        和Tool类注释中给出的示例大体上是一样的，区别在于Tool类中示例使用了JobConf类直接控制MR任务的配置，并使用JobClient类，以JobConf为配置参数直接提交任务；而Grep类示例中，使用Job类,似乎是封装更严实，提供的选项更多些：

//设置并构置Hadoop环境
Configuration conf = getConf();
Job grepJob = Job.getInstance(conf);
grepJob.setJobName("grep-search");
grepJob.setJarByClass(Grep.class);
//配置MapReduce环境和任务
FileInputFormat.setInputPaths(grepJob, args[0]);
grepJob.setMapperClass(RegexMapper.class);
grepJob.setCombinerClass(LongSumReducer.class);
grepJob.setReducerClass(LongSumReducer.class);
FileOutputFormat.setOutputPath(grepJob, tempDir);
grepJob.setOutputFormatClass(SequenceFileOutputFormat.class);
grepJob.setOutputKeyClass(Text.class);
grepJob.setOutputValueClass(LongWritable.class);
//启动并等待任务
grepJob.waitForCompletion(true);

        实际上，Grep是在Job.waitForCompletion(true)中,调用了Job.submit();在Job.submit()中，调用了JobSubmitter.submitJobInternal(Job.this,cluster)。

        而在Tool的示例中，是使用静态方法JobClient.runJob(JobConf job)，以传入的Hadoop配置环境conf为参数新建一个JobClient对象，并以该对象调用了JobClient类的submitJob方法，其中又调用了JobClient.submitJobInternal(conf)；在submitJobInternal中则新建Job对象，并调用了重写的Job.run()方法；该重写方法主要作用是跳过MR配置过程——因为已经配置过了，直接使用Job.submit()提交MR任务，最终也还是归于JobSubmitter.submitJobInternal(Job.this,cluster)。

        输入输出方面，需要注意的是JobConf并没有实现setInputPath方法，但是在JobConf的注释里给出了对应的示例，使用的是FileInputFormat的setInputPaths方法，这个就和Grep示例一样了，内部都是调用的Configuration.set方法。

        总而言之——

        简单的理解，开始一个MR任务的主要动作，其实都是在配置Configuration的内部参数，手段是通过Configuration的set方法、setClass方法、setJar方法等等。配置完成后，以配置好的Configuration为参数创建一个Job对象，submit就行了。

        5.ToolRunner类        

 
public class ToolRunner {
public static int run(Configuration conf, Tool tool, String[] args) 
    throws Exception{
    if(conf == null) {
      conf = new Configuration();
    }
    GenericOptionsParser parser = new GenericOptionsParser(conf, args);
    //set the configuration back, so that Tool can configure itself
    tool.setConf(conf)
    String[] toolArgs = parser.getRemainingArgs();
    return tool.run(toolArgs);
  }
   public static int run(Tool tool, String[] args) 
    throws Exception{
    return run(tool.getConf(), tool, args);
  }
  
  public static void printGenericCommandUsage(PrintStream out) {
    GenericOptionsParser.printGenericCommandUsage(out);
  }
  
}

        ToolRunner是一个封装类，主要目的是提供了一个静态run函数，用于启动实现tool.run接口的mapreduce任务，通常该run函数在main方法中调用。函数包含3个参数：

        参数1：Configuration conf是整个mapreduce的运行环境封装

        参数2：Tool tool是Mapreduce的算法流程封装

        参数3：args argstring输入输出。

        概要代码如下：

    If(conf==nullptr)conf = new Configuration();
    tool.setConf(conf);
    tool.run(args);

        6.RegexMapper

        Grep使用MapReduce提供的正则搜索Mapper算法RegexMapper，对输入文件进行分片处理并按行匹配模板。RegexMapper类继承自Mapper类。

        Mapper作为模板类共4个参数，用来定义输入KV对和输出KV对的数据类型：

        参数1，2：KEYIN，VALUEIN

        参数3，4：KEYOUT，VALUEOUT

        Mapper内部还嵌入了一个Context抽象类，该类用于管理Mapper算法内部的数据处理，比如：①输出Mapper结果时，是调用Context的write方法；②枚举需要Mapper的输入时，调用的是Context的getCurrentKey、getCurrentValue和nextKeyvalue方法。

        实际上，实现任何一个Map算法都应该从继承Mapper类开始，并重写（override）Mapper的setup、cleanup、map、run方法。

        方法1：setup（Context context），在初始化时调用一次，可用于设置全局参数，比如在RegexMapper中，此处用于初始化pattern、group这2个私有变量的值，其具体取值是通过Grep在run方法中调用conf.set(RegexMapper.PATTERN,args[2])和conf.set(RegexMapper.GROUP,args[3])设置在常量中的。

        方法2：cleanup(Context context)，用于在对象销毁时提供清理资源的机会

        方法3：map(KEYIN key,VALUEIN value,Context context)，用于定制map方法。每次map被调用时，实际都是一个键值对被传输进来，定制用户需要根据输入key，value值来确定输出键值对，并通过Context.write方法输出。比如mapper中内置了这个默认实现，直接将输入的key，value通过Context.write给写了出去。

        方法4：run(Context context)。run方法代码如下，可见当mapper类被加载时，其主要工作就是从context中枚举被拆分的键值对，然后调用map方法。

{
    setup(context);
    try {
        while (context.nextKeyValue()) {
            map(context.getCurrentKey(), context.getCurrentValue(), context);
        }
    }
    finally {
      cleanup(context);
    }
}

        7.TextInputFormat/FileInputFormat/InputFormat

        在示例代码中，第一次MR操作仅仅对输出的键值对数据类型做了配置，没有对输入进行配置。事实上，在未显式设置的情况下，MapReduce默认使用TextInputFormat作为输入类型，即输入文件为文本文件，输入键值对为类型，其中Text为文本文件中的每一行，为TextInputFormat中的createRecordReader方法通过实例化LineRecordReader构造。

        TextInputFormat类定义为public abstract class TextInputFormat extends FileInputFormat，可见默认情况下其键值对的类型已明确定义，且TextInputForamt继承自FileInputFormat。

        继承自FileInputFormat类定义为Public abstract class FileInputFormat extends InputFormat，即该类继承自InputFormat。

        TextInputFormat主要重写了3个方法：

        方法1：RecordReader createRecordReader(InputWritable split, TaskAttemptContext context);通过实例化LineRecordReader类，将根据JobSubmitDir中的信息读取的文件分块进一步切分为键值对。

        方法2：protected Boolean isSplitable(JobContext context,Path file)，用来指示通过JobSubmit提交的splite是否需要进一步切分。

FileInputFormat主要重写了1个方法：

        方法3：public List getSplits(JobContext job)；主要用于将大文件切分为MapReduce适宜调度的块。该方法将在工作调度程序中使用。

InputFormat是一个抽象类，主要定义了需要被重写的initialize、getSplits、createRecordReader方法。

        8.LineRecordReader/RecordReader

        RecordReader是用于维护管理键值对的抽象类，主要定义了5个方法。

        方法1：initialize（InputSplit，TaskAttemptContext）；在初始化时调用一次，可在这里实现对split的进一步拆分，构建键值对列表

        方法2：nextKeyValue()；移动到下一个键值对；

        方法3：KEYIN getCurrentKey();得到当前键值对的键；

        方法4：VALUEIN getCurrentValue();得到当前键值对的值；

        方法5：float getProgress();得到当前RecordReader的处理进度（0.0-1.0）

        LineRecordReader继承了recordReader，重写了initialize、nextKeyValue、getCurrentKey、getCurrentValue和getProgress等方法。值得提一下的是，LineRecordReader似乎并未在initialize中直接将split完全拆开后保存，而是仅构建了维护信息，读取了首个键值对。其余键值对仅在nextkeyValue被调用时通过in.readLine(value,……)临时读出。这应该是针对海量文件批处理的特点考虑，毕竟都放在内存里面可能就爆了。

        RecordReader的5个方法，和Context所实现的TaskAttemptContext接口有明显的意义对应关系，容易猜到，在mapper的run方法中，context对应的接口，应该是调用了RecordReader的方法。这个实际上是在MapContextImpl类的nextKeyValue中实现的——MapContextImpl实现了Context规定的接口。

        9.MapReduce的mapper的总体执行流程

        前面提到，整个MapReduce的工作流程都是在JobTracer和TaskTracer的管理下运转的。事实上，这也是人云亦云的结果。我们在3.3.4版本的Single Node中没有观察到这2个进程。当然，这可能是版本（原描述来自于2.0版本）和模式（Single Node不可能出现多进程）的结果。然而我们还是选择相信，因为即使在SingleNode模式下的多线程方式，其工作流程和人云的方式也是非常相似的。至于Multi模式就不尝试了，毕竟我们的目的只是大致了解一下MR的工作流程而已，Single Node模式的流程如下图所示。也由于不求甚解的原因，部分代码并没有深究，所以不能保证完全正确：

         图中，实线箭头代表进程执行流程和函数调用关系；虚线代表参数关联关系；空心箭头代表线程迁移关系；更细的空心箭头则代表数据在被处理过程中的形态变化。

        Grep调用时，InputFormat的具体类型是TextInputFormat，其父类是FileInputFormat，此时getSplits的作用是按照Hadoop默认的块大小切分文件。

        JobSplitWriter.createSplitFiles将切分的文件的文件名、偏移等信息写回文件系统，代码中称其为JobSubmitDir，临时文件位置在/tmp/Hadoop/mapred/staging/下。

        然后，在JobSubmitter中开始正式提交工作，使用的是submit name

        调用submitClient.submitJob()后就看不到代码了，下一次命中是在Local Job Runner Map Task Executor线程中。

        MapTask.run()，这里读取JobSubmitDir，并根据任务状态（setup/cleanup/runnewmapper或者runoldmapper）执行动作。如调用runNewMapper函数，构建mapper；进一步调用NewTrackingRecordReader方法，其中会调用具体输入类TextInputFormat的createRecordReader方法，基于LineRecordReader类，在split的基础上，再次拆分split为RecordReader。

        然后，MapTask会真正的启动Map任务：

        Input.initialize(split,mapperContext);

        Mapper.run(mapperContext);

        在Mapper.run中对RecordReader拆分的键值对进行处理。