Hadoop是怎么分块的

http://blog.chinaunix.net/space.php?uid=20602285&do=blog&cuid=2273160

hadoop的分块有两部分,其中第一部分更为人熟知一点。
 
第一部分就是数据的划分(即把File划分成Block),这个是物理上真真实实的进行了划分,数据文件上传到HDFS里的时候,需要划分成一块一块,每块的大小由hadoop-default.xml里配置选项进行划分。
 
<property>
  <name>dfs.block.size</name>
  <value>67108864</value>
  <description>The default block size for new files.</description>
</property>
 
这个就是默认的每个块64MB。
数据划分的时候有冗余,个数是由
<property>
  <name>dfs.replication</name>
  <value>3</value>
  <description>Default block replication.
  The actual number of replications can be specified when the file is created.
  The default is used if replication is not specified in create time.
  </description>
</property>
指定的。
 
具体的物理划分步骤要看Namenode,这里要说的是更有意思的hadoop中的第二种划分。
 
在hadoop中第二种划分是由InputFormat这个接口来定义的,其中有个getSplits方法。这里就有了一个新的不为人熟知的概念:Split。Split的作用是什么,Split和Block是什么关系,下面就可以说明清楚。
在Hadoop0.1中,split划分是在JobTracker端完成的,发生在JobInitThread对JobInProgress调用inittasks()的时候;而在0.18.3中是由JobClient完成的,JobClient划分好后,把split.file写入hdfs里,到时候jobtracker端只需要读这个文件,就知道Split是怎么划分的了。
第二种划分只是一种逻辑上划分,目的是为了让Map Task更好的获取数据输入,仔细分析如下这个场景:
 
File 1 : Block11, Block 12, Block 13, Block 14, Block 15
File 2 : Block21, Block 22, Block 23
 
File1有5个Block,最后一个Block当然可能小于64MB;File2有3个Block
 
如果用户在程序中指定map tasks的个数,比如说是2(如果不指定的话maptasks个数默认是1),那么在
FileInputFormat(最常见的InputFormat实现)的getSplits方法中,首先会计算totalSize=8(可以对照源码看看,注意getSplits这个函数里的计量单位是Block个数,而不是Byte个数,后面有个变量叫bytesremaining仍然表示剩余的Block个数,有些变量名让人无语),然后会计算goalSize=totalSize/numSplits=4,对于File1,计算一个Split有多少个Block是这样计算的
 
long splitSize = computeSplitSize(goalSize, minSize, blockSize);
protected long computeSplitSize(long goalSize, long minSize, long blockSize) {
 return Math.max(minSize, Math.min(goalSize, blockSize));
}
这里minSize是1(说明了一个Split至少包含一个Block,不会出现一个Split包含零点几个Block的情况),计算得出splitSize=4,所以接下来Split划分是这样分的:
Split 1: Block11, Block12, Block13,Block14
Split 2: Block15
Split 3: Block21, Block22, Block23
那用户指定的map个数是2,出现了三个split怎么办?在JobInProgress里其实maptasks的个数是根据Splits的长度来指定的,所以用户指定的map个数只是个参考。可以参看JobInProgress: initTasks()
里的代码:
 
  try {
   splits = JobClient.readSplitFile(splitFile);
  } finally {
   splitFile.close();
  }
  numMapTasks = splits.length;
  maps = new TaskInProgress[numMapTasks];
 
所以问题就很清晰了,还如果用户指定了20个map作业,那么最后会有8个Split(每个Split一个Block),所以最后实际上就有8个MapTasks,也就是说maptask的个数是由splits的长度决定的。
 
几个简单的结论:
1. 一个split不会包含零点几或者几点几个Block,一定是包含大于等于1个整数个Block
2. 一个split不会包含两个File的Block,不会跨越File边界
3. split和Block的关系是一对多的关系
4. maptasks的个数最终决定于splits的长度
 
 
还有一点需要说明,在FileSplit类中,有一项是private String[] hosts;
看上去是说明这个FileSplit是放在哪些机器上的,实际上hosts里只是存储了一个Block的冗余机器列表。
比如上面例子中的Split 1: Block11, Block12, Block13,Block14,这个FileSplit中的hosts里最终存储的是Block11本身和其冗余所在的机器列表,也就是说Block12,Block13,Block14存在哪些机器上没有在FileSplit中记录。
 
FileSplit中的这个属性有利于调度作业时候的数据本地性问题。如果一个tasktracker前来索取task,jobtracker就会找个task给他,找到一个maptask,得先看这个task的输入的FileSplit里hosts是否包含tasktracker所在机器,也就是判断和该tasktracker同时存在一个机器上的datanode是否拥有FileSplit中某个Block的备份。
 
但总之,只能牵就一个Block,其他Block就从网络上传罢。

============================================================================

http://www.geminikwok.com/2011/07/26/hadoop是怎么分块的/

hadoop的分块有两部分。

第一部分就是数据的划分(即把File划分成Block),这个是物理上的划分,数据文件上传到HDFS里的时候,需要划分成一块一块,每块的大小由hadoop-default.xml里配置选项进行划分(大小不足一块时,便按实际大小存放):

<property>

<name>dfs.block.size</name>

<value>67108864</value>

<description>The default block size for new files.</description></property>

这里设置的是每个块64MB。
数据划分的时候也可以设置备份的份数:
<property>

<name>dfs.replication</name>

<value>3</value>

<description>Default block replication.   The actual number of replications can be specified when the file is created.  The default is used if replication is not specified in create time.  </description>

</property>
具体的物理划分步骤由Namenode决定,下面hadoop中的第二种划分,用来决定M/R运行时,一个map处理的数据量。

在hadoop中第二种划分是由InputFormat这个接口来定义的,其中有个getSplits方法。这里有一个新的概念:fileSplit。每个map处理一个fileSplit,所以有多少个fileSplit就有多少个map(map数并不是单纯的由用户设置决定的)。

我们来看一下hadoop分配splits的源码:

if ((length != 0) && isSplitable(fs, path)) {

long blockSize = file.getBlockSize();

long splitSize = computeSplitSize(goalSize, minSize, blockSize);

long bytesRemaining = length;

while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {

int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);

splits.add(new FileSplit(path, length-bytesRemaining, splitSize, blkLocations[blkIndex].getHosts()));

bytesRemaining -= splitSize;        }

if (bytesRemaining != 0) {

splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining,  blkLocations[blkLocations.length-1].getHosts()));        }

} else if (length != 0) {

splits.add(new FileSplit(path, 0, length,blkLocations[0].getHosts()));

} else {

//Create empty hosts array for zero length files

splits.add(new FileSplit(path, 0, length, new String[0]));

}

}

从代码可以看出,一个块为一个splits,即一个map,只要搞清楚一个块的大小,就能计算出运行时的map数。而一个split的大小是由goalSize, minSize, blockSize这三个值决定的。computeSplitSize的逻辑是,先从goalSize和blockSize两个值中选出最小的那个(比如一般不设置map数,这时blockSize为当前文件的块size,而goalSize是文件大小除以用户设置的map数得到的,如果没设置的话,默认是1),在默认的大多数情况下,blockSize比较小。然后再取bloceSize和minSize中最大的那个。而minSize如果不通过”mapred.min.split.size”设置的话(”mapred.min.split.size”默认为0),minSize为1,这样得出的一个splits的size就是blockSize,即一个块一个map,有多少块就有多少map。

上面说的是splitable的情况,unsplitable可以根据实际情况来计算,一般为一个文件一个map。

下面是摘自网上的一个总结:

几个简单的结论:
1. 一个split不会包含零点几或者几点几个Block,一定是包含大于等于1个整数个Block
2. 一个split不会包含两个File的Block,不会跨越File边界
3. split和Block的关系是一对多的关系
4. maptasks的个数最终决定于splits的长度

还有一点需要说明,在FileSplit类中,有一项是private String[] hosts;
看上去是说明这个FileSplit是放在哪些机器上的,实际上hosts里只是存储了一个Block的冗余机器列表。
比如有个fileSplit 有4个block: Block11, Block12, Block13,Block14,这个FileSplit中的hosts里最终存储的是Block11本身和其备份所在的机器列表,也就是说 Block12,Block13,Block14存在哪些机器上没有在FileSplit中记录。

FileSplit中的这个属性有利于调度作业时候的数据本地性问题。如果一个tasktracker前来索取task,jobtracker就会找个 task给他,找到一个maptask,得先看这个task的输入的FileSplit里hosts是否包含tasktracker所在机器,也就是判断 和该tasktracker同时存在一个机器上的datanode是否拥有FileSplit中某个Block的备份。

但总之,只能牵就一个Block,其他Block就要从网络上传。不过对于默认大多数情况下的一个block对应一个map,可以通过修改hosts使map的本地化数更多一些。 在讲block的hosts传给fileSplit时,hosts中的主机地址可以有多个,表示map可以从优先从这些hosts中选取(只是优先,但hdfs还很可能根据当时的网络负载选择不是hosts中的主机起map task)。

知道这个特性之后,可以修改传回给fileSplit的hosts,在列表中只写block所在的那些hosts,这样hdfs就会优先将这些map放到这些hosts上去执行,由于hosts上有该block,就省掉了网络传输数据的时间。

这样做的话,在job很多的时候,可能会出现hot spot,即数据用的越多,它所在hosts上的map task就会越多。所以在考虑修改传给fileSplit的时候要考虑平衡诸多因素

原文地址:https://www.cnblogs.com/xuxm2007/p/2162011.html