杰霸霸博客

主题

InputFormat数据输入 ​

1. 切片与MapTask并行度决定机制 ​

  1. 问题引出
    MapTask的并行度决定Map阶段的任务处理并发度，进而影响到整个Job的处理速度。 思考：1G的数据，启动8个MapTask，可以提高集群的并发处理能力。那么1K的数据，也启动8个MapTask，会提高集群性能吗？MapTask并行任务是否越多越好呢？哪些因素影响了MapTask并行度？
  2. MapTask并行度决定机制
    数据块：Block是HDFS物理上把数据分成一块一块。数据块是HDFS存储数据单位。 数据切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。数据切片是MapReduce程序计算输入数据的单位，一个切片会对应启动一个MapTask。 Alt text

2. Job提交流程源码和切片源码详解 ​

Hadoop源代码执行流程梳理：

java
waitForCompletion()
// 1建立连接
    connect();    
        // 1)创建提交Job的代理
        new Cluster(getConfiguration());
            // (1)判断是本地运行环境还是yarn集群运行环境
            initialize(jobTrackAddr, conf); 

// 2 提交job
submitter.submitJobInternal(Job.this, cluster)

    // 1)创建给集群提交数据的Stag路径
    Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);

    // 2)获取jobid ，并创建Job路径
    JobID jobId = submitClient.getNewJobID();

    // 3)拷贝jar包到集群
copyAndConfigureFiles(job, submitJobDir);    
    rUploader.uploadFiles(job, jobSubmitDir);

    // 4)计算切片，生成切片规划文件
writeSplits(job, submitJobDir);
        maps = writeNewSplits(job, jobSubmitDir);
        input.getSplits(job);

    // 5)向Stag路径写XML配置文件
writeConf(conf, submitJobFile);
    conf.writeXml(out);

    // 6)提交Job,返回提交状态
status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());

2.1 job提交流程源码解析 ​

Alt text

2.2 FileInputFormat切片源码解析(input.getSplits(job)) ​

  1. 程序先找到数据存储的目录
  2. 开始遍历该目录下的每个文件
  3. 遍历第一个ss.txt
    a. 获取文件大小fs.sizeOf(ss.txt) b. 计算切片大小
    computeSplitSize(Math.max(minSize, Math.min(maxSize, blocksize)))=blocksize=128M
    c. 默认情况下，切片大小=blocksize
    d. 开始切，形成第一个切片：ss.txt--0:128M 第二个切片ss.txt--128:256M 第三个切片ss.txt--256:300M
    (每次切片时，都要判断剩下的部分是否大于块的1.1倍，不大于1.1倍就划分一块切片)
    e. 将切片信息写到一个切片规划文件中
    f. 整个切片的核心过程在 getSplit() 方法中完成
    g. InputSplit只记录了切片的元数据信息，比如起始位置、长度以及所在的节点列表等。
  4. 提交切片规划文件到YARN上, YARN上的MrAppMaster就可以根据切片规划文件计算开启MapTask个数。

2.3 源码中计算切片大小的公式 ​

切片大小=Math.max(minSize, Math.min(maxSize, blockSize))
其中minSize=mapreduce.input.fileinputformat.split.minSize 默认值为1
maxSize= mapreduce.input.fileinputformat.split.maxSize 默认值为Long.MaxValue
因此默认情况下，切片大小=blocksize。

2.4 切片大小的设置 ​

  • 切片大小调小： 修改maxsize(切片最大值)，如果参数调的比blockSize小，则会让切片变小。
  • 切片大小调大： 修改minSize(切片最小值)，如果参数调的比blockSize大，则会让切片变大。

2.5 切片信息相关API ​

java
// 获取切片的大小名称
String name = inputSplit.getPath().getName();
// 根据文件类型获取切片信息
FileSplit inputSplit = (FileSplit) context.getInputSplit();

3. FileInputFormat切片机制 ​

3.1 切片机制 ​

  • 简单地按照文件的内容长度进行切片
  • 切片大小，默认等于Block大小
  • 切片时不考虑数据集整体，而是逐个针对每一个文件单独切片

举例分析

  1. 输入数据有两个文件：
    file1.txt 320M
    file2.txt 10M
  2. 经过FileInputFormat的切片机制运算后，形成的切片信息如下：
    file1.txt.split1-- 0~128M
    file1.txt.split2-- 128~256M
    file1.txt.split3-- 256~320M
    file2.txt.split1-- 0~10M

4. FileInputForma实现类 ​

思考：在运行MapReduce程序时，输入的文件格式包括：基于行的日志文件、二进制格式文件、数据库表等。那么，针对不同的数据类型，MapReduce是如何读取这些数据的呢？
FileInputFormat常见的接口实现类包括：TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定义InputFormat等。

5. TextInputFormat ​

TextInputFormat是默认的FileInputFormat实现类。按行读取每条记录。键是存储该行在整个文件中的起始字节偏移量, LongWritable类型。值是这行的内容，不包括任何行终止符(换行符和回车符)，Text类型。

6. CombineTextInputFormat ​

框架默认的TextInputFormat切片机制是对任务按文件规划切片，不管文件多小，都会是一个单独的切片，都会交给一个MapTask，这样如果有大量小文件，就会产生大量的MapTask，处理效率极其低下。

6.1 应用场景： ​

CombineTextInputFormat用于小文件过多的场景，它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个MapTask处理。

6.2 虚拟存储切片最大值设置 ​

java
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m

警告

虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。

6.3 切片机制 ​

生成切片过程包括：虚拟存储过程和切片过程二部分。

6.3.1 虚拟存储过程 ​

将输入目录下所有文件大小，依次和设置的setMaxInputSplitSize值比较，如果不大于设置的最大值，逻辑上划分一个块。如果输入文件大于设置的最大值且大于两倍，那么以最大值切割一块；当剩余数据大小超过设置的最大值且不大于最大值2倍，此时将文件均分成2个虚拟存储块(防止出现太小切片)。
例如setMaxInputSplitSize值为4M，输入文件大小为8.02M，则先逻辑上分成一个4M。剩余的大小为4.02M，如果按照4M逻辑划分，就会出现0.02M的小的虚拟存储文件，所以将剩余的4.02M文件切分成(2.01M和2.01M)两个文件。

6.3.2 切片过程 ​

  1. 判断虚拟存储的文件大小是否大于setMaxInputSplitSize值，大于等于则单独形成一个切片。
  2. 如果不大于则跟下一个虚拟存储文件进行合并，共同形成一个切片。
  3. 测试举例: 有4个小文件大小分别为1.7M、5.1M、3.4M以及6.8M这四个小文件，则虚拟存储之后形成6个文件块,
    大小分别为: 1.7M，(2.55M、2.55M)，3.4M以及(3.4M、3.4M)
    最终会形成3个切片，大小分别为: (1.7+2.55)M，(2.55+3.4)M，(3.4+3.4)M
    切片举例
  4. 代码实操:
    • 不做任何处理，运行前面的WordCount案例程序，观察切片个数为4。 Alt text
    • 在WordcountDriver中增加如下代码，运行程序，并观察运行的切片个数为3。
      (a)驱动类中添加代码如下:
java
// 如果不设置InputFormat，它默认用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);

//虚拟存储切片最大值设置4m
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);
    (b)运行如果为3个切片:

Alt text

  1. 在WordcountDriver中增加如下代码，运行程序，并观察运行的切片个数为1。
    • 驱动中添加代码如下:
java
// 如果不设置InputFormat，它默认用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);
//虚拟存储切片最大值设置20m
CombineTextInputFormat.setMaxInputSplitSize(job, 20971520);

运行结果： Alt text