zoukankan      html  css  js  c++  java
  • Hadoop基础【1.3】多路径输入 指定Map 自定义输入分片

    1. 多路径输入

    对于一个MR程序来说,再所有情况下都只使用一个Path作为我们的输入路径是不现实的,一般都需要从多个路径下取文件。因此提供了FileInputFormat下的static方法setInputPaths以实现此功能(此例中多个路径使用同一个Mapper)。

    1 String inPath = "root";
    2 String[] inputPathStr = new String[]{inPath+"/a",inPath+"/b",inPath+"/c",inPath+"/d",inPath+"/e"};
    3 Path[] inPaths = new Path[inputPathStr.length];int i=0;
    4 for(String inPathStr:inputPathStr){
    5     inPaths[i++] = new Path(inPathStr);         
    6 }
    7 FileInputFormat.setInputPaths(job,inPaths);

    2. 多路径输入各自对应Mapper

    当多路径输入输入数据集时,若文件数据格式不同,则无法用同一个Mapper来操作,可以通过MultipleInputs将每个Path对应一个Mapper以解决此问题。

    1 job.setJarByClass(xxxxxMapper.class);
    2 
    3 MultipleInputs.addInputPath(jon,new Path("pathA"),TextInputFormat.class,xxxxxxxMapper.MapA.class);
    4 
    5 MultipleInputs.addInputPath(jon,new Path("pathB"),TextInputFormat.class,xxxxxxxMapper.MapB.class);
    6 
    7 job.setReduceClass(xxxxxReducer.class);

    再自定义的Mapper里多写几个map,然后再MultipleInputs通过addInputPath()进行各自数据Map方法的设置

    3. 自定义输入分片

    问题1: 分片、map任务数、reduce任务数关系

    Ans:一般情况下,在输入源是文件的时候,一个task的map数量由splitSize来决定的,如5M的文件,SqlitSize为1M,5片(SplitSize计算在1.2里写到了)。当输入数据为数据库时,也可通过jobconf.set(“mapred.map.tasks.nums”,100)进行设置。

    reduce在运行时往往需要从相关map端复制数据到reduce节点来处理,因此相比于map任务。reduce节点资源是相对比较缺少的,同时相对运行较慢,正确的reduce任务的个数应该是0.95或者1.75 *(节点数 ×mapred.tasktracker.tasks.maximum参数值)。如果任务数是节点个数的0.95倍,那么所有的reduce任务能够在 map任务的输出传输结束后同时开始运行。如果任务数是节点个数的1.75倍,那么高速的节点会在完成他们第一批reduce任务计算之后开始计算第二批 reduce任务,这样的情况更有利于负载均衡。同时需要注意增加reduce的数量虽然会增加系统的资源开销,但是可以改善负载匀衡,降低任务失败带来的负面影响。同样,Reduce任务也能够与 map任务一样,通过设定JobConf 的conf.setNumReduceTasks(int num)方法来增加任务个数。

    问题2:对小文件如何MR

    Ans:当对大量小文件进行MR时,一个小文件一个MapTask,效率过低。通过CombineFileInputFormat将多个“小文件”合并为一个"切片",而后return <K,V>输入map,减少maptask数量,增大效率。

    第一次:将同DN上的所有block生成Split,生成方式:

      1.循环nodeToBlocks,获得每个DN上有哪些block

      2.循环这些block列表

      3.将block从blockToNodes中移除,避免同一个block被包含在多个split中

      4.将该block添加到一个有效block的列表中,这个列表主要是保留哪些block已经从blockToNodes中被移除了,方便后面恢复到blockToNodes中

      5.向临时变量curSplitSize增加block的大小

      6.判断curSplitSize是否已经超过了设置的maxSize

        a) 如果超过,执行并添加split信息,并重置curSplitSize和validBlocks

        b) 没有超过,继续循环block列表,跳到第2步

      7.当前DN上的block列表循环完成,判断剩余的block是否允许被split(剩下的block大小之和是否大于每个DN的最小split大小)

        a) 如果允许,执行并添加split信息

        b) 如果不被允许,将这些剩余的block归还blockToNodes

      8.重置

      9.跳到步骤1

    第二次:对不再同一个DN上但是在同一个机架上的block进行合并(只是之前还剩下的block)

    最后: 对于既不在同DN也不在同机架的block进行合并(经过前两步还剩下的block)

    https://blog.csdn.net/wawmg/article/details/17095125

    http://blog.itpub.net/31506529/viewspace-2217548/

    在自定义的InputFormat里的createRecordRecord方法返回 new CombineFileRecordReader<K,V>((CombineFileSplit)inputSplit,context,MyXxxxRecordReader),对于自定义的RecordReader,构造函数,initialize,nextKeyValue比较重要注意一下。

    问题3:自定义InputFormat

    Ans:啊一般不需要自定义getSplits的情况下,继承了FileInputFormat后,重写createRecordReader就好了,RecordReader在map里主要的就是nextKeyValue(),因此会调用我们指定的RecordReader中的nextKeyValue函数。这个函数就会处理或者说是初始化key和value,然后返回true,告知已经处理好了。接着就会调用getCurrentKey 和getCurrentValue获取当前的key和value值。最后,返回map,继续执行map逻辑。

    https://blog.csdn.net/u011007180/article/details/53310209

  • 相关阅读:
    STL目录
    Hola!
    SWPUCTF 2019总结以及部分WP
    SQL手工注入基础篇
    JDK11,JDK12没有JRE的解决方法
    FJUT2019暑假周赛三部分题解
    FJUT2019暑假周赛一题解
    随笔1
    关于针对本校教务系统漏洞的一次信息检索
    KMP算法讲解
  • 原文地址:https://www.cnblogs.com/tillnight1996/p/12359669.html
Copyright © 2011-2022 走看看