hadoop hdfs总结 NameNode部分-- 概述

转载地址为：http://www.cnblogs.com/sidmeng/archive/2012/03.html

搞hadoop一段时间了，总的来说一些东西都是零零总总，没有形成一个系统总结一下，在今后的blog中，总结相关内容是将会是接下来的内容。
先从概述来讲一下hadoop hdfs的结构，hdfs由四部分组成，分别是1、Namenode(包括有INode，blockMap，FSNamesystem,FSDirectory等结构)；2、Datanode(Datanode,FSDataset等);3、Client(DFSClient,DistributeFileSystem等) 4、底层的IPC通信机制。

其中Namenode，Datanode以及Client都是通过IPC定义相关的protocol进行通信，内部逻辑之间紧密相连。

之后会从这四个方面一一介绍。

以后的代码总结全部基于cloudera CDH3B2源码进行。

NameNode需要处理三部分工作：

1、NameNode需要对内存中的数据进行管理，包括有INode信息，Block信息，其中INode信息是间断性的向磁盘刷，进行固话，而Block信息则是在启动过程中从Datanode的心跳汇报中获得的。在这个过程中，还需要对日志进行处理，包括FSImage和Editlog。

2、与DataNode进行通信。

3、与Client进行通信。

其中2、3是通过底层的IPC框架完成，2、3将会在后面对DataNode和Client进行介绍时详细说明。

hdfs的设计参照了linux ext文件系统的设计，也有INode，Block等概念。在NameNode中INode是元数据信息(metadata)，同时由于其分布式特点，必须包含了Block对应的DataNode信息以及DataNode所有包含的Block信息，即对应的BlockMap。同时，NameNode的日志保存了元数据信息。

hadoop hdfs总结 NameNode部分 1

一、INode部分

先画了一下类图，比较简单。



1、INode是抽象类，有两个子类，INodeFile和INodeDirectory，对了了file和directory。

INode定义了一些基本属性，如name,parent,modificationTime,accessTime，还有ugi信息等。INode implements Comparable是可比较的，可以通过二分查找找到树状结构中对应的INode。

此外，INode中重要的方法是public final ContentSummary computeContentSummary() ，计算该INode树状结构下的统计数据，如file数，dir数，length等。

INode中还有一个重要方法 static byte[][] getPathComponents(String[] strings) 该方法的作用是获取文件名称，按照seperate 即"/"，每个path存在一个byte[]数组中。

2、INodeDirectory定义了Directory，基本属性有private List<INode> children 是所有树状结构底下的INodeFile和INodeDirectory，INodeDirectory的一些重要方法：

（1）INode removeChild(INode node) 通过二分查找找到对应的子INode对象，从children中删除，并返回该对象

（2）void replaceChild(INode newChild) 同上，只不过是替换

（3）private INode getChildINode(byte[] name) 通过二分查找获得INode对象

（4）<T extends INode> T addChild(final T node, boolean inheritPermission) 插入一个INode对象到队列中，并且按顺序插入，其中重要的代码是

?
（5）<T extends INode> INodeDirectory addToParent 插入一个INode到相关的parent中，并且更新该parent对应的INode children队列

（6)DirCounts spaceConsumedInTree(DirCounts counts) 计算磁盘使用空间，递归计算

(7)int collectSubtreeBlocksAndClear(List<Block> v) 将所有树状结构下的block返回，用于删除以后对block做处理。

（8）int getExistingPathINodes(byte[][] components, INode[] existing) 这个方法个人认为比较重要，作用是从components[]中获取存在的INode数组，意思是components是一个全路径path，有可能存在该路径，也可能不存在，通过该函数找到存在的path，形成INode对象返回

int getExistingPathINodes(byte[][] components, INode[] existing) {
assert compareBytes(this.name, components[0]) == 0 :
"Incorrect name " + getLocalName() + " expected " + components[0];

INode curNode = this;
int count = 0;
int index = existing.length - components.length;
if (index > 0)
index = 0;
while ((count < components.length) && (curNode != null)) {
if (index >= 0)
existing[index] = curNode;
if (!curNode.isDirectory() || (count == components.length - 1))
break; // no more child, stop here
INodeDirectory parentDir = (INodeDirectory)curNode;
curNode = parentDir.getChildINode(components[count + 1]);
count += 1;
index += 1;
}
return count;
}

3、INodeDirectoryWithQuota 继承自INodeDirectory，加入了Quota,如果超出Quota会抛出相应的ExceededException 异常。

4、INodeFile INodeFile对应了Block，HDFS对每个File复制了多分副本，通过策略放在不同的DataNode上，同时每个INodeFile对应了多个Block,默认情况下每个Block64M。

INodeFile中主要的属性有protected BlockInfo blocks[]，protected short blockReplication，protected long preferredBlockSize

blocks在blockMap中会详细介绍，就是该INodeFile对应的Block。

INodeFile中的一些重要方法：

（1）Block getLastBlock() 获得最后一个Block，由于目前HDFS支持append，所以获得最后一个Block进行append操作

（2）void addBlock(BlockInfo newblock) 向该INodeFile对应的Block数组中最后位置添加一个Block

（3）int collectSubtreeBlocksAndClear(List<Block> v) 将所有Block放在列表中，如删除某个目录下所有文件，就需要获得所有Block，在BlockMap中进行删除。

?
设置parent=null 使得资源得以通过GC回收。

（4）Block getPenultimateBlock() 获得倒数第二个Block

5、INodeFileUnderConstruction 说明该INodeFile对应的Block正在写入DataNode。INodeFile处于UnderConstruction状态有几种情况，如正写入文件，lease recovery等。当然只能对最后一个Block进行append。

主要属性有 String clientName; 正在写入的Client名称。 private int primaryNodeIndex = -1; 当进行leaseRecovery时，会选出primary通过primary向其它DataNode发送Recovery信息。

private DatanodeDescriptor[] targets = null; target是最后一个Block对应的DataNode位置，该target由一定算法获得。在后面会分析。

重要的方法有：

（1）void setTargets(DatanodeDescriptor[] targets) 设置该INodeFile对应Block所在的DataNode(DatanodeDescriptor是DataNode的抽象）

（2）void addTarget(DatanodeDescriptor node) 将新的DataNode加入到target中

（3）INodeFile convertToInodeFile() 如果DataNode获得所有数据并成功回报给NameNode后，将INodeFIleUnderConstruction状态转化为正常状态。

(4) void removeBlock(Block oldblock) 将最后一个Block删除，原因可能是由于Block传输过程中出错，被abandon掉，也可能是进行recovery中发现时间戳不一致。

（5）synchronized void setLastBlock(BlockInfo newblock, DatanodeDescriptor[] newtargets） 设置最后一个Block,并且设置对应的DataNode，在NameNode的Append中使用。

（6）void assignPrimaryDatanode() 在LeaseRecovery过程中使用，到LeaseRecovery过程会详细解释，这个方法就是找到一个alive的DataNode作为primary，来进行Lease Recovery。

INode部分至此结束，INode+BlockMap构成了NameNode中主要的数据结构，其它的操作都是围绕这这两个数据结构进行处理，后面将会总结BlockMap。

hadoop hdfs总结 NameNode部分 2

二、BlocksMap部分

在NameNode 介绍1中，说明了INode部分，INode抽象了NameNode中文件属性，与之相同的是Block部分，Block是对NameNode中实际存储的Block的抽象。这两部分构成了NameNode的最重要的元信息。

Block继承结构为基类Block,BlockInfo继承自Block,是BlocksMap的内部类，BlocksMap是BlockInfo的一个hashmap存储。

首先说明一下BlockInfo的意义，在hdfs中，数据实际上是存储在DataNodes中的Block，每个Block是64M(可配置)，文件较大时对应需要多个Block进行存储，同时为了安全性每个Block进行了多份备份，存储在不同的DataNodes中(NameNode中算法分配)。INodeFile中有protected BlockInfo blocks[] 表示该INodeFile由几个Block组成。而BlockInfo内部结构则表示了该Block备份到了哪几个DataNodes中，同时BlockInfo还有一个三元组结构，类似于数据结构中的双向链表，除了保存该Block存储在哪个DataNode中还保存了该DataNode存储的其他BlockInfo引用。引入三元组的原因是除了要得到INodeFile--->Blocks的映射，和Block--->DataNodes的映射外，还需要得到DataNodes--->Blocks的映射。一旦DataNodes故障或者下线，能够将其对应的所有Block进行备份。

详细说明各个结构：

1、Block

Block需要在NameNode，DataNodes间进行传输，所以进行了序列化。主要属性为

?
　　其他方法都是围绕着三个属性进行。在DataNodes中block存储为blk_$blockId的形式。generationStamp是为了进行append过程或recovery过程设置的time stamp。

2、BlockInfo

BlockInfo是BlocksMap的内部类，继承了Block类。属性为

?
　　inode是该BlockInfo对应的INodeFile,triplets就是前面说过的三元组。它表示为该BlockInfo被复制在多个DataNodes中，结构为triplets[3*i]表示同一个Block复制到的某个DataNode，triplets[3*i+1]和triplets[3*i+2]表示该DataNodes对应的其它Block，可以认为是前驱和后继Block。DataNodeDescriptor有一个头结点的引用，这个会在后面分析说到。

一些重要的方法是：

(1)DatanodeDescriptor getDatanode(int index)

获得第i个DataNode。

(2)BlockInfo getPrevious(int index) BlockInfo getNext(int index)

获得第i个DataNode的前驱BlockInfo和后继BlockInfo。

(3)void setPrevious(int index, BlockInfo to) void setNext(int index, BlockInfo to)

设置该第i个DataNode对应该BlockInfo的前驱和后继BlockInfo结点。

(4)private int getCapacity()

该Block存储在多少个DataNode中。

(5)int numNodes()

获得该Block对应的有效的DataNodes个数。

(6)private int ensureCapacity(int num)

加入新的DataNodes，如setReplication由3变为7，需要将block复制到其他DataNodes节点中。主要代码为

?
　　由于DataNodes个数不会太多，所以这种复制效率是可以接受的。

(7)BlockInfo listInsert(BlockInfo head, DatanodeDescriptor dn)

DataNodeDerscriptor中维护了一个头结点，将this引用加入到list中。

(8)BlockInfo listRemove(BlockInfo head, DatanodeDescriptor dn)

将this引用从DatanodeDescriptor中的链表(三元组)中删除。

3、BlocksMap

BlocksMap将BlockInfo wrapper。

?
　 重要方法是：

(1)

?
　　检查map是否存在该BlockInfo,若没有则create一个BlockInfo，put 到map中。

(2)

?
　　添加特定BlockInfo的INode对象。

(3)

?
　　删除该BlockInfo对象。删除涉及到几个结构，首先是map中，然后是从DatanodeDescriptor中删除对应的Block。

hadoop hdfs总结 NameNode部分 3 ----DatanodeDescriptor

DatanodeDescriptor是对DataNode的抽象，它是NameNode的内部数据结构，配合BlockMap和INode，记录了文件系统中所有Datanodes包含的Block信息，以及对应的INode信息。

DatanodeDescriptor继承自DatanodeInfo,DatanodeInfo继承自DatanodeID。

一、DatanodeID

DatanodeID有以下属性：

public String name; /// hostname:portNumber

public String storageID; /// unique per cluster storageID 集群内唯一的hostname

protected int infoPort; /// the port where the infoserver is running infoPort的端口号

public int ipcPort; /// the port where the ipc server is running 底层IPC通信端口号

二、DatanodeInfo

1、DatanodeInfo有以下属性：

protected long capacity;

protected long dfsUsed;

protected long remaining;

protected String hostName = null; hostname由Datanode在register时候提供

protected long lastUpdate;

protected int xceiverCount; 这个比较重要，表示的是Datanode与client或者Datanode连接时候的连接数，超出后会出错

protected String location = NetworkTopology.DEFAULT_RACK; 网络拓扑结构，这个可以定义，按照机架进行备份放置策略

protected AdminStates adminState; adminState表示的是此Datanode的运行状态，运行状态有NORMAL, DECOMMISSION_INPROGRESS, DECOMMISSIONED; 在Datanode进行decommission时候有用，decommission指的是Datanode下线，为了防止数据丢失，在下线过程中需要将此Datanode对应的Block拷贝到其他Datanode上。

2、重要方法

public String dumpDatanode() 将所有的属性统计信息输出。

三、DatanodeDescriptor

DatanodeDescriptor是对DataNode所有操作的抽象，DataNode就是存储文件系统的所有数据，数据对应了文件，文件由多个块构成，每个块又有多分备份。对于DataNode的操作，基本上有client向Datanode传输数据，Datanode需要记录所有的block，如果数据丢失需要将block进行重新复制(replicate)，如果数据在append过程或者传输过程中产生错误，需要进行恢复(recovery)等等。DatanodeDescriptor中封装了所有的操作。

1、重要数据结构

(1)内部类 BlockTargetPair

public static class BlockTargetPair {
public final Block block;
public final DatanodeDescriptor[] targets;

BlockTargetPair(Block block, DatanodeDescriptor[] targets) {
this.block = block;
this.targets = targets;
}
}

表示的是block以及对应所有副本存放的Datanode。为下面的一些数据结构提供基础。

(2)内部类private static class BlockQueue

用来对BlockTargetPair队列进行封装，包括出列入列等方法。

(3)private volatile BlockInfo blockList = null;

每个DatanodeDescriptor要记录该Datanode所保存的所有Block，就是通过BlockInfo来保存的，blockList根据三元组存储(见BlocksMap分析)，作为头节点，存储所有的block，通过链表来获得。

(4)内部结构：

/** A queue of blocks to be replicated by this datanode */
private BlockQueue replicateBlocks = new BlockQueue();
/** A queue of blocks to be recovered by this datanode */
private BlockQueue recoverBlocks = new BlockQueue();
/** A set of blocks to be invalidated by this datanode */
private Set<Block> invalidateBlocks = new TreeSet<Block>();

这些内部结构包括有需要由这个Datanode复制给其它Datanode的----replicateBlock,需要由该Datanode复制给其它Datanode的----recoverBlocks，需要将Block从Datanode删除的。

前两个结构需要得到其它DatanodeDescriptor，由于需要获知需要进行复制和恢复的Datanode，而invalidate只是本次Datanode需要删除的，与其它Datanode无关。

(5)以下变量维护了block调度包括block report和heartbeat时间等。

private int currApproxBlocksScheduled = 0;
private int prevApproxBlocksScheduled = 0;
private long lastBlocksScheduledRollTime = 0;
private static final int BLOCKS_SCHEDULED_ROLL_INTERVAL = 600*1000; //10min

2、重要方法

(1)void updateHeartbeat

void updateHeartbeat(long capacity, long dfsUsed, long remaining,
int xceiverCount) {
this.capacity = capacity;
this.dfsUsed = dfsUsed;
this.remaining = remaining;
this.lastUpdate = System.currentTimeMillis();
this.xceiverCount = xceiverCount;
rollBlocksScheduled(lastUpdate);
}

DataNode向NameNode进行心跳汇报时，更新状态，包括有capacity，dfsused，remainning和xceiverCount,并且将最后更新时间更新。

(2)boolean addBlock(BlockInfo b)

boolean addBlock(BlockInfo b) {
if(!b.addNode(this))
return false;
// add to the head of the data-node list
blockList = b.listInsert(blockList, this);
return true;
}

将block插入到队列头。

(3)boolean removeBlock(BlockInfo b)

boolean removeBlock(BlockInfo b) {
blockList = b.listRemove(blockList, this);
return b.removeNode(this);
}

从队列中删除。

(4)void addBlockToBeReplicated

void addBlockToBeReplicated(Block block, DatanodeDescriptor[] targets) {
assert(block != null && targets != null && targets.length > 0);
replicateBlocks.offer(block, targets);
}

将Block放置在replicateBlocks结构中。

(5)void addBlockToBeRecovered

void addBlockToBeRecovered(Block block, DatanodeDescriptor[] targets) {
assert(block != null && targets != null && targets.length > 0);
recoverBlocks.offer(block, targets);
}

将Block放置在recoverBlocks结构中。

(6)void addBlocksToBeInvalidated

void addBlocksToBeInvalidated(List<Block> blocklist) {
assert(blocklist != null && blocklist.size() > 0);
synchronized (invalidateBlocks) {
for(Block blk : blocklist) {
invalidateBlocks.add(blk);
}
}
}

将Block放置在invalidateBlocks结构中。

(7) BlockCommand getReplicationCommand(int maxTransfers)

BlockCommand getLeaseRecoveryCommand(int maxTransfers)

BlockCommand getInvalidateBlocks(int maxblocks)

这三个方法相同，就是将三个内部数据结构中的数据封装成writable的数据形式传输给对应的Datanode，同时将cmd指定为DatanodeProtocol.DNA_TRANSFER，DatanodeProtocol.DNA_RECOVERBLOCK或者DatanodeProtocol.DNA_INVALIDATE。

(8)reportDiff 这个方法是DatanodeDescriptor中最重要的方法

void reportDiff(BlocksMap blocksMap,
BlockListAsLongs newReport,
Collection<Block> toAdd,
Collection<Block> toRemove,
Collection<Block> toInvalidate) {
// place a deilimiter in the list which separates blocks
// that have been reported from those that have not
BlockInfo delimiter = new BlockInfo(new Block(), 1);
boolean added = this.addBlock(delimiter);
assert added : "Delimiting block cannot be present in the node";
if(newReport == null)
newReport = new BlockListAsLongs( new long[0]);
// scan the report and collect newly reported blocks
// Note we are taking special precaution to limit tmp blocks allocated
// as part this block report - which why block list is stored as longs
Block iblk = new Block(); // a fixed new'ed block to be reused with index i
for (int i = 0; i < newReport.getNumberOfBlocks(); ++i) {
iblk.set(newReport.getBlockId(i), newReport.getBlockLen(i),
newReport.getBlockGenStamp(i));
BlockInfo storedBlock = blocksMap.getStoredBlock(iblk);
if(storedBlock == null) {
// If block is not in blocksMap it does not belong to any file
toInvalidate.add(new Block(iblk));
continue;
}
if(storedBlock.findDatanode(this) < 0) {// Known block, but not on the DN
// if the size differs from what is in the blockmap, then return
// the new block. addStoredBlock will then pick up the right size of this
// block and will update the block object in the BlocksMap
if (storedBlock.getNumBytes() != iblk.getNumBytes()) {
toAdd.add(new Block(iblk));
} else {
toAdd.add(storedBlock);
}
continue;
}
// move block to the head of the list
this.moveBlockToHead(storedBlock);
}
// collect blocks that have not been reported
// all of them are next to the delimiter
Iterator<Block> it = new BlockIterator(delimiter.getNext(0), this);
while(it.hasNext())
toRemove.add(it.next());
this.removeBlock(delimiter);
}

Datanode会定期向NameNode进行report，当然由于report十分消耗资源，所有report时间不会非常频繁。当汇报时候，会将新获得的Block与BlocksMap中的Block进行对比，如果BlocksMap中不存在该Block，则删除。如果缺少副本数则添加，其它的加入道Datanode到Block的映射中。

Hadoop-FSDataset

FSDataset
FSDir
1、getGenerationStampFromFile(File[]listdir, File blockFile)
获取时间戳，data存取在Datanode上的规则是blk_blockId,metadata存储在blk_blockId_timeStamp.meta
2、getBlockInfo(TreeSet<Block>blockSet)
获取block对应的info包括File，length，和timestamp
3、getBlockAndFileInfo(TreeSet<BlockAndFile>blockSet)
与上面相同，不同是获得绝对路径
4、getVolumeMap(HashMap<Block,DatanodeBlockInfo>
volumeMap, FSVolume volume)
把所有文件放到一个map中
5、checkDirTree()
检查dir状态
6、clearPath(Filef,
String[] dirNames, intidx)
目前看就是减少block数，在上层会有解决
FSVolume
FSVolume有以下属性：
privateFile
currentDir; //保存到的目录
privateFSDir
dataDir; //目前的目录对应的FSDir
privateFile
tmpDir;
privateFile
blocksBeingWritten;
// clients write here
privateFile
detachDir;// copy on write for blocks in snapshot
privateDF
usage;
privateDU
dfsUsage;
privatelongreserved;
//为系统其它部分留取的空间
在其初始化的阶段对各个文件夹初始化，对于blocksBeingWritten目录如果其支持append操作，那么其中的文件不能够被删除。
1、addBlock(Blockb,
File f)
把tmp文件夹下已经写完的文件放到current文件夹下
2、recoverBlocksBeingWritten(Filebbw)
当dn挂掉时，之前正在写的文件
FSVolumeSet
保存了FSVlume的set，如果在conf中设置了多个文件，就对应了多个FSVolume
ActiveFile类
文件写的时候会进入active状态，ActiveFile类就是针对这种情况进行处理
1、FilefindMetaFile(finalFile
blockFile)
获取metafile
2、longparseGenerationStamp(File
blockFile, File metaFile)
获取timeStamp
3、synchronized
File findBlockFile(longblockId)
获得block文件，锁机制解决
4、synchronizedBlock
getStoredBlock(longblkid)
获得block，同上，但是获得的时block
5、MetaDataInputStreamgetMetaDataInputStream(Block
b)
获得metadata的输入流，这个用来验证等
FSDataSet类的成员有
FSVolumeSet
volumes;
privateHashMap<Block,ActiveFile>
ongoingCreates=
newHashMap<Block,ActiveFile>();
privateintmaxBlocksPerDir=
0;
HashMap<Block,DatanodeBlockInfo>volumeMap =newHashMap<Block,
DatanodeBlockInfo>();;
static Random
random=
newRandom();
privateintvalidVolsRequired;
FSDatasetAsyncDiskServiceasyncDiskService;
记住一条block等已经进行了序列化，可以在网络中进行传输，包含了所有的block信息
1、synchronizedlonggetVisibleLength(Block
b)
获得block的length，如果正在些，就获得目前已经能够看到的length。
2、synchronizedFile
getBlockFile(Block b)
获得block对应的file
3、synchronizedInputStream
getBlockInputStream(Block b, longseekOffset)
获得对应block的file的InputStream，最终返回的是FileInputStream
4、synchronized
BlockInputStreamsgetTmpInputStreams(Block b,

longblkOffset,
longckoff)
返回的时block和checksum的封装的inputStream
5、BlockWriteStreamscreateBlockWriteStreams(
File f , File metafile)
获得向硬盘些的输出流，一个封装
6、updateBlock(Blockoldblock,
Block newblock)
synchronizedList<Thread> tryUpdateBlock(Block
oldblock, Block newblock)
首先获得该block是否对应了正在写的线程。如果没有则update当前的block状态。
7、writeToBlock(Blockb,
booleanisRecovery,
booleanreplicationRequest)
这是很重要的一个方法。对于isRecovery的情况，有一种时由于block已经写入Dn但是有些ack没有返回的时候dn挂掉了，重启后会重新向NN汇报。另外的情况就是append操作。
1）首先会检查该block是否已经存在，并且可以被recovery，进行recovery。
2）如果该block存在，但是正在被创建中，并且不是append的操作则抛出异常，由Datanode捕捉异常，再打log。表示该block已经存在了
3）如果该block存在，且正在append，则停止所有append的进程。
4）接着对append操作进行处理，首先获得原来的block信息，将原有的meta更换为现在的meta，将block和meta放到tmp目录中，同时防盗volumeMap中
5）最后生成一个BlockWriteStreams的流，进行写操作
8、booleandelBlockFromDisk(File
blockFile, File metaFile, Block b)
删除block和meta数据
9、Block[]getBlockReport()
获得所有的block信息

hadoop 集群间数据迁移

hadoop集群之间有时候需要将数据进行迁移，如将一些保存的过期文档放置在一个小集群中进行保存。

使用的是社区提供的功能，distcp。用法非常简单：

hadoop distcp hdfs://nn1:8020/foo/bar hdfs://nn2:8020/bar/foo

加上参数 -i 表示不用去管failure -m 设置map数

hadoop mapreduce学习 Client部分

Mapreduce framework 同hdfs架构有类似的部分，分为JobTracker(对应Namenode)，TaskTracker（对应Datanode），Job（对应DFSClient)。功能虽然不太相同，但是原理还接近。

总结一下Job的流程。Job是对JobClient的封装，Job本身功能比较简单，无非是获得各种参数，将参数封装好，向JobTracker提交job的过程。但是JobClient内部还是有些流程的。

1、在提交Job时候，首先JobClient向JobTracker要求获得一个unique的ID，即JobID

JobID jobId = jobSubmitClient.getNewJobId();

2、然后调用copyAndConfigureFiles函数，将所有需要的文件，以及第三方的lib都考入到distribuedCache中。同时将mapreduce需要操作的jar文件拷到底层的hdfs中，拷贝10份。

Path submitJarFile = JobSubmissionFiles.getJobJar(submitJobDir);
job.setJar(submitJarFile.toString());
fs.copyFromLocalFile(new Path(originalJarPath), submitJarFile);
fs.setReplication(submitJarFile, replication);
fs.setPermission(submitJarFile,
new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION));

3、下一步就需要对input path中也就是说需要处理的数据进行split(切分)，每一个split对应一个map task，每个split又包含了许多的record（由key-value）组成。split只是逻辑上的概念，并不是真正的数据split，split结束确定每个split的位置(把BlockLocation对象包了一层)，以及split的个数。比如FileInputFormat类中对File进行了split，过程为：

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;
}

split结束后需要将split的元数据(即split切分结果）写入到hdfs中。

public static <T extends InputSplit> void createSplitFiles(Path jobSubmitDir,
Configuration conf, FileSystem fs, T[] splits)
throws IOException, InterruptedException {
FSDataOutputStream out = createFile(fs,
JobSubmissionFiles.getJobSplitFile(jobSubmitDir), conf);
SplitMetaInfo[] info = writeNewSplits(conf, splits, out);
out.close();
writeJobSplitMetaInfo(fs,JobSubmissionFiles.getJobSplitMetaFile(jobSubmitDir),
new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION), splitVersion,
info);
}

至此split结束，返回值为切分个数，即map task个数。

4、最后将conf.xml拷贝到hdfs中，向jobTrakcer提交该Job，至此Job才真正提交。返回一个NetworkedJob，表明一个提交了的Job，通过该类来获取已经提交Job的所有信息。

所有的job 提交的jar file 和xml，以及job.split和job.splitmetainfo都存放在staging文件夹下，我存放的位置是/user/hadoop/.staging/ 需要有权限才能查看。其中jar file和job.split file都是有10个副本。

总结一下：在Job提交之前，所有需要拷贝的文件都已经拷贝到dfs中，split已经切分好。

使用NFS服务提高hadoop 可靠性

众所周知 hadoop是一个单点系统，即所谓的spof，所以元数据的保护就是重中之重。目前元数据写两份，分别是本地的磁盘，远程的磁盘。还有一个secondary namenode做冷备。写到远程的磁盘就需要的是NFS服务了。

NFS(Network File System, 网络文件系统)可以通过网络将分享不同主机(不同的OS)的目录——可以通过NFS挂载远程主机的目录, 访问该目录就像访问本地目录一样。

首先下载NFS,网址是http://sourceforge.net/projects/nfs/files/ 选择合适的版本下载。到server和client机器安装。

编辑server主机的文件/etc/exports /etc/hosts.allow /etc/hosts.deny三个文件。

对/etc/exports 加入例如：/data0 10.39.2.121(rw,no_root_squash)

其中/data0为对外开放的目录 后面是机器加权限，详细说明见
http://nfs.sourceforge.net/nfs-howto/ar01s03.html

启动nfs server的服务：

/etc/init.d/portmap start

/etc/init.d/nfslock start

/etc/init.d/nfs start

在client端同样启动相同的服务。

同时在client端挂在远程目录 sudo mount master:/data0/ /data1

/data1是client端本地目录。

nfslock 导致 namenode无法format

接挂载nfs硬盘后，重新启动namenode，以便所有元数据能够顺利写到nfs硬盘中。过程为首先namenode 发现nfs硬盘没有format，先进行format在将fsimage 写入nfs中。在启动过程中发现如下错误：

ERROR org.apache.hadoop.hdfs.server.common.Storage: Cannot create lock on /×××/in_use.lock

java.io.IOException: No locks available

at sun.nio.ch.FileChannelImpl.lock0(Native Method)

at sun.nio.ch.FileChannelImpl.tryLock(FileChannelImpl.java:871)

at java.nio.channels.FileChannel.tryLock(FileChannel.java:962)

at org.apache.hadoop.hdfs.server.common.Storage$StorageDirectory.tryLock(Storage.java:627)

at org.apache.hadoop.hdfs.server.common.Storage$StorageDirectory.lock(Storage.java:605)

at org.apache.hadoop.hdfs.server.common.Storage$StorageDirectory.analyzeStorage(Storage.java:463)

at org.apache.hadoop.hdfs.server.namenode.FSImage.recoverTransitionRead(FSImage.java:300)

at org.apache.hadoop.hdfs.server.namenode.FSDirectory.loadFSImage(FSDirectory.java:99)

at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.initialize(FSNamesystem.java:358)

at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.<init>(FSNamesystem.java:327)

at org.apache.hadoop.hdfs.server.namenode.NameNode.initialize(NameNode.java:271)

at org.apache.hadoop.hdfs.server.namenode.NameNode.<init>(NameNode.java:465)

at org.apache.hadoop.hdfs.server.namenode.NameNode.createNameNode(NameNode.java:1224)

at org.apache.hadoop.hdfs.server.namenode.NameNode.main(NameNode.java:1233)

错误原因，nfslock服务启动有问题。解决方法在nfs的客户端和服务器端都运行sudo service nfslock start或者sudo service nfslock restart