MapReduce task任务

maptask的并行度：map阶段的分布式运行

整个map阶段的任务被分成了几个小任务,运行每一个小任务就称为一个maptask任务，一个maptask对应的是计算的任务量
注意：一个maptask任务只能运行在一个节点上，一个节点上可以运行多个maptask任务的
整个任务被分成几个maptask任务，取决于节点的数据量。

maptask的任务数的条件

对4个文件进行统计

  • 大小1.42kb
  • 3个从节点 datanode nodemanager

在进行 MapReduce任务的时候，发现在 Map阶段执行 jps 命令的时候，可以看到控制台出现4个yarnchild。yarnchild—–maptask任务—-mapper类
1个yarnchild—–1个maptask任务

问题：maptask任务和数据有关的? 还是和数据的文件个数有关？

假设文件个数

假设和数据的文件个数有关，那么4份1T数据：4maptask 分别处理 1T的数据，

  • 假设有10个节点，最多运行在4个节点上，那么会剩下6个节点没有事情做 – 负债不均衡

  • 1）负载不均衡

    2）数据存储是分块存储(128M的块大小小) 1t—8192块 分散存储在10个节点上
    每一个maptask要访问8192个数据块 ，跨节点访问数据不合理

maptask任务和文件的个数无关
maptask任务应该数据量有关

数据量有关

一个maptask任务对应多少数据量？

1
2
3
4
5
6
假设wc---1个文件 500M
实际存储：
    blk01:0-127M
    blk02:128-255M
    blk03:256-383M
    blk04:384-500M

  • 假设对应100M数据量—一个maptask任务：处理100M的数据

    1
    2
    3
    4
    maptask01:0-99M    blk01
    maptask02:100-199M  blk01  blk02
    maptask03:200-299M  blk02  blk03
    .......

    有极大的可能跨数据块,跨节点访问数据:性能低 不合理

  • 对应的数据量200M—一个maptask任务 处理200M数据量

    1
    2
    maptask01:0-199M     blk01  blk02
    .......

    ​ 有极大的可能跨数据块 跨节点访问数据:不合理

  • 分析 maptask对应的数据量128M的时候最合理的

    1
    2
    3
    4
    5
    maptask01:blk01
    maptask02:blk02
    maptask03:blk03
    maptask04:blk04
    底层实现是否是这样的？

MapTask的并行度

事实上一个maptask任务对应的数据量是一个切片大小—split

切片

逻辑上的概念（逻辑切块） 逻辑切片，仅仅是一个偏移量的划分

  • 一个maptask任务对应数据量就是一个逻辑切片
  • 一个逻辑切片假设100M 意思就是这个逻辑切片对应的数据0-99M数据 不会真正的切分数据
  • 理论上切片的大小就是数据块的大小

切片大小- getSplits()

文件输入类：FileInputFormat 的getSplits方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
    //...
    long blockSize = file.getBlockSize();
    long splitSize = this.computeSplitSize(goalSize, minSize, blockSize);

    //....
}

//通过分析代码
//获取最终的切片大小
//goalSize--128M
//minSize--1
//blockSize--long_max
protected long computeSplitSize(long goalSize, long minSize, long blockSize) {
    return Math.max(minSize, Math.min(goalSize, blockSize));
    //推导：
    //Math.min(maxSize====long_max, blockSize==128*1024*1024M)    ====blocksize
    //Math.max(minSize===1,blocksize===128M)   ========   blocksize
    //得到的结果就是blocksize=128M
}

修改Split的大小

1
2
3
修改切片大小：
小于128M-----修改max属性
大于128M-----修改min属性

一、 修改配置文件

1
2
3
4
1. max---
   mapreduce.input.fileinputformat.split.maxsize
2. min----
   mapreduce.input.fileinputformat.split.minsize

二、修改代码

1
FileInputFormat.setMaxInputSplitSize(job, 1*1024);

总结

maptask的并行度 和切片大小有关
一个切片—-1个maptask任务
默认的时候1个切片的大小就是一个数据块的大小

maptask任务在哪一个节点执行由yarn资源分配决定的
maptask任务 执行MyMapper的类的代码

block split的区别？

  • block：hdfs的数据存储的切分单元 默认128M(hadoop1.x是64M) 物理切分
  • split：mapreduce中maptask任务的时候 进行数据划分的单元 逻辑上的概念 没有真正的切分数据
    理论上默认的split的大小===blocksize
    当前的文件 不够128M 则单独成一个逻辑切片
    最后一个切片 最大大小 blockSize/splitSize*1.1 128M*1.1=140.8M