hadoop hdfs文件系统分析，名字节点，数据节点之间的交互

MapReduce提供Partitioner接口，它的做用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪一个reduce task处理。
2.map task与reduce task的执行是否在不一样的节点上？
3.Shuffle产生的意义是什么？
4.每一个map task都有一个内存缓冲区，存储着map的输出结果，当缓冲区快满的时候须要将缓冲区的数据该如何处理？
5.在map task执行时，它是如何读取HDFS的？
6.读取的Split与block的对应关系多是什么？
7.MapReduce提供Partitioner接口，它的做用是什么？
8.溢写是在什么状况下发生？
9.溢写是为何不影响往缓冲区写map结果的线程？
10.当溢写线程启动后，须要对这80MB空间内的key作排序(Sort)。排序是MapReduce模型默认的行为，这里的排序也是对谁的排序？
11.哪些场景才能使用Combiner呢？

若是client设置过Combiner，那么如今就是使用Combiner的时候了。将有相同key的key/value对的value加起来，减小溢写到磁盘的数据量。Combiner会优化MapReduce的中间结果，因此它在整个模型中会屡次使用。那哪些场景才能使用Combiner呢？从这里分析，Combiner的输出是Reducer的输入，Combiner毫不能改变最终的计算结果。因此从个人想法来看，Combiner只应该用于那种Reduce的输入key/value与输出key/value类型彻底一致，且不影响最终结果的场景。好比累加，最大值等。Combiner的使用必定得慎重，若是用好，它对job执行效率有帮助，反之会影响reduce的最终结果。
12.Merge的做用是什么？
13.reduce中Copy过程采用是什么协议？
14.reduce中merge过程有几种方式，与map有什么类似之处？
15.溢写过程当中若是有不少个key/value对须要发送到某个reduce端去，那么如何处理这些key/value值

问题导读：
1、什么是MapReduce？
2、Mapper、Reducer实现什么工做内容？
3、如何进行MapReduce性能调优？

hadoop学习

google file ystem Doug cutting等在Nutch项目上应用gfs和mapreduce思想，演化出hadoop项目，

提到了lucene和nutch项目,lucene是引擎开发工具包，nutch还有数据抓取功能，

狭义的hadoop包括hadoop common,hadoop hdfs和hadoop mapreduce三个子项目，可是想相关的还包括Avro,ZooKeeper,Hive,Pig,HBase,还有其余互补的软件项目。

zookeeper用来解决分布式计算中的一致性问题。

hadoop配置信息处理

配置文件，

在操做系统中，有ini配置文件，里面有节[],项和值

在java properties文件中没有分类节，继承自hashtable;

Hadoop Configuration 管理配置文件：xml格式；根元素是configuration,通常只包含子元素property,每一个property元素就是一个配置项

合并资源是将多个配置文件合并，并产生一个配置；经过Configuration类的addResources()方法，合并成一个配置文件。

hadoop配置系统还有一个属性扩展${hadoop.tmp.dir}/dfs/name, ${hadoop.tmp.dir}会用Configuration中的相应属性进行扩展

使用configuaration的通常过程是:用addResource()方法添加须要加载的资源，而后经过get,set….对资源进行操做。

getProps方法中，采用了延迟加载的设计模式，当真正须要配置数据的时候，才开始分析配置文件

配置文件有windows的ini配置文件，有java的properties文件，properties文件也能够用方法保存为xml,也能够从xml中读成properties，j2se1.5版本之后，Properties中的数据也能够用xml格式保存，对应的加载和写入方法分别是loadFomXML和storeToXML(),

hadoop使用dom处理xml文件，由于文件小。能够所有加载到内存后，再进行解析。

Configurable接口位于org.apache.hadoop.conf包中

序列化与压缩

对象的序列化（Serialization）是用于将对象编码成一个字节流，以及从字节流中从新构建对象。“将一个对象编码成一个字节流”称为序列化该对象；相反的过程称为反序列化。

文本->对象序列化

对象->文本反序列化

序列化主要有三种用途：

做为一种持久化的格式

做为一种通讯数据格式，序列化结果能够从一个正在运行的虚拟机，经过网络被传输到另外一个虚拟机上。

做为拷贝和克隆的机制，到内存缓存区，获得新拷贝的对象

Serializable接口是一个标志，不具备任何成员函数

public interface Serializable{}

ObjectOutputStream对象，对java自定义类，和基本类型的序列化

ObjectInputStream对象

java序列化机制

hadoop序列化机制

hadoop writable机制

RawCompatator和WritableComparable接口

hadoop序列化框架

Avro

Thrift

Google Protocol Buffer

Hadoop提供了一个简单的序列化框架API,集成各类序列化的实现，该框架由 Serialization实现(org.apache.hadoop.io.serializer包中)

数据压缩每每是计算密集型的操做，考虑到性能，建议使用本地库，Java本地方法就是一个不错的选择，java提供了一些钩子程序，使得调用本地方法成为了可能，本地方法是类的成员方法。

org.apace.hadoop.io序列化和压缩

Hadoop远程过程调用

远程调用如何隐藏底层通讯的细节

xml-rpc json-rpc

hadoop-ipc

常规的过程调用，归纳了进程的控制流同步模型

RPC跨越了不一样的进程

theClinet and theServer

请求参数保存在栈中，调用方法，从栈中取得参数，肯定调用过程当中的名字并调用相应的过程，调用结束后，返回值经过主程序打包并发送给客户端，通知客户端调用结束

实现RPC须要关心下面三个问题

RPC的语法应该与高级程序设计语言中的本地过程调用语法有相同的外观；

RPC和本地调用的语义尽量相似，如参数传递中包含的语义

Rpc接收者应该如同一个传统的调用。

RPC引入了客户存根和服务器骨架。

服务器骨架和客户端存根是等价物，用来将经过网络的输入请求转为本地调用。调用结束后，由骨架打包成消息发送到客户端存根，存根经过return将结果返回到主程序中，

RPC经过接口定义语言IDL描述调用接口的相关信息，定义信息包含类型定义，常量声明，参数传递信息和注释。

java远程方法调用是一个Java的一个核心API和类库。容许一个java虚拟机上运行的java程序调用不一样虚拟机上运行的对象中的方法，RMI能够当作是RPC的java升级版。包含RMI的java程序包含服务程序和客户端程序，服务器应用程序将建立多个远程对象，使这些对象可以被客户端引用。标准的Stub/Skeleton机制

Stub表明被客户端引用的远程对象，并保存着远程对象的接口和方法列表，位于客户端；服务端的Skeleton对象处理有关调用“远方”对象中的全部细节。

public class RMIQueryStatusImpl extends UnicastRemoteObject implements RMIQueryStatus

客户端要访问远程对象必须得到一个本地的存根对象。

如何得到存根对象呢？java中用RMI远程对象注册点(Registry)解决了这个问题

一个Rmi典型的存根包含哪些成员。

Java动态代理

java.lang.reflect包下，主要包含java.lang.relflect.Proxy和java.lang.reflect.InvocationHandler

反射，代理

反射，便于编写可以动态操做java代码的程序

函数seek()来自org.apache.hadoop.fs.Seekable接口，提供在输入流中进行定位的能力。

“cre/0”是校验文件的魔数(Magic NUmber)

Arrays.equals(a,b)

ChecksumException

数据完整性是海量数据处理中的一个重要问题。不但hadoop Common中的文件系统提供了ChecksumFileSystem,在hadoop分布式文件系统中也实现了相似的机制。

RawLocalFileSystem和ChecksumFileSystem并无实现一个真正的文件系统，必须依赖一个已有的文件系统。

RawInMemoryFileSystem不依赖任何文件系统，十分简单，内存文件系统RawInMemeryFileSystem

内存文件系统它部分实现了抽象类org.apache.hadoop.fs.FileSystem中要求的API及简单的输入/输出流。

InMemoryFileSystem内部类RawInMemoryFileSystem,内存文件系统没有目录，全部文件都以映射表的形式保存在系统中，映射表的键是文件名，映射表对应的值是FileAttribues对象，它包含预留空间和该空间大小的信息；这样的映射表有两个，一个保存正在产生的文件tempFileAttribs中，经过映射表这样的文件结构将数据保存在了内存中。

public class FileAttributes {

private byte[] data;// 保存数据

private int size; // 保存大小

public FileAttributes(int size) {

this.size = size;

this.data = new byte[size];

}

HDFS

超大文件，硬件故障，流式数据访问，简化一致性，一次写入，屡次读取，简化可一致性模型，提升了吞吐量；大量的小文件会影响名字节点的性能。

HDFS体系结构

主从:名字节点，数据节点，客户端Client

数据块：块的大小表明系统读写操做的最小单位，Linux的Ext3文件系统中，块大小默认为4096字节

默认HDFS数据块的大小是64mb,减小须要管理数据块须要的开销，减小创建网络链接须要的成本。

名字节点维护着整个文件系统的文件目录树，文件/目录的元信息和文件的数据块索引，即每一个文件对应的数据块列表；

这些信息以两种形式存储在本地文件系统中：一种是命名空间镜像FSImage,另外一种是命名空间镜像的编辑日志(EditLog).

空间镜像保存着某一特定时刻HDFS的目录树，元信息和数据块索引等信息，对这些信息的改动，则保存在编辑日志中；

第二名字节点用于按期合并命名空间镜像和镜像编辑日志的辅助守护进程，它不接收或者记录HDFS的任何实时变化，而是间隔不停的得到HDFS某一时间点的命名空间镜像和镜像的编辑日志，合并获得一个新的命名空间镜像，该镜像会上传到名字节点，替换原有的命名空间镜像，并清除上述日志。应该说，第二名字节点配合名字节点，为名字节点上的名字节点第一关系提供一个检查点机制Checkpoint，避免编辑日志过大，致使名字节点启动时间过长的问题。

第二名字节点能够减小停机的时间并减低名字节点元数据丢失的风险，名字节点失效不行。

数据节点上都会有一个守护进程，将HDFS数据块写到linux本地文件系统的实际文件中，或者从实际文件中读取数据库。

数据节点会和其余数据节点通讯，复制数据块，保证数据的冗余性。

客户端是HDFS和用户交互的手段，JAVAAPI，命令行，其余类库等等。。

hdfs.web

hdfs.server.namenode.web.resources hdfs.server.datanode.web.resources

hdfs.web.resources

hdfs.tools hdfs.server.balancer

hdfs hdfs.server.namenode hdfs.server.datanode

hdfs.protocal hdfs.server.common

hdfs.server.namenode.metrics hdfs.server.datanode.metrics

hdfs.util hdfs.security.token.bloacl hdfs.security.token.do….

hadoop远程调用接口

基于tcp和http的流式接口

版本号用来维护具备相同数据块标识的数据的一致性

数据块的id号<id>, name为blk_<id>

LocatedBlock已经确认存储位置的数据块

数据节点标识datanodeID用于在HDFS中肯定一个数据节点，

经过create()建立文件时，ClientProtocal的成员函数addBlock()用于为当前客户端打开的文件添加一个数据块。当用户建立一个新文件，或者写满一个数据块后，客户端就会调用这个方法，向名字节点申请一个新的数据块。

在linux中，大多数磁盘IO操做都会经过内核缓存区，若是缓存区还没有写满，则不会发生实际的io操做，这种机制叫作延迟写，它的优势是减小了磁盘io操做的次数，但缺点是一旦系统出故障，可能会丢失那些在缓存区但未被写进磁盘的文件数据。

public class BlockCommand extends DatanodeCommand{

Block blocks[]; //包含了blockCommand命令具体涉及的数据块

DatanodeInfo targets[][];//是一个类型为DatanodeInfo的二维数组，用于数据块复制和数据块恢复

}

namenodeProtocal

调用者有两个，一个是第二名字节点，另外一个是hdfs工具：均衡器balancer

balancer能够把数据块从改数据节点移动到其余数据节点，达到平衡各个数据节点数据块数量的目的；

第二名字节点的功能：获取hdfs命名空间镜像和镜像编辑日志，合并到一个新的镜像，上传新命名空间镜像到名字节点，替换原有镜像并清空镜像编辑日志。

hdfs写数据的数据流管道：

GFS写一份数据的多个副本时，能够充分利用集群中每一台机器的宽带。

首先推送到一个节点，而后再向下一个节点推送。

流式接口：

fsimage,edits文件

hadoop的各个节点都嵌入了Jetty,是一个web容器，实现了java的http servlet规范，

用http协议来交换数据，

${dfs.data.dir}

数据节点的实现

一个目录下最多只有64个数据块(128个文件)和64个目录。

数据存储DataStorage和文件系统数据集FSDataset.

StorageDirectory

Storage

DataStorage

数据节点第一次启动的时候，会调用DataStorage.format()建立存储目录结果。

Storage和DataStorage一块儿提供了一个完美的HDFS数据节点升级机制，简化了大型分布式系统的运维，它们不但解决了系统升级过程当中数据格式的转化,升级回滚等常见的问题，并保证了这个过程不会丢失数据；同时，特别是Storage状态机，以及配合状态机工做的临时文件，提供了完整的升级方案。在升级过程或者回滚的过程当中任何一个步骤出现错误，均可以经过状态机，恢复到正常状态。

文件系统数据集的工做机制

FSVolume是数据目录配置项${dfs.data.dir}中的一项。

ongoingCreates是文件块Block对象到ActiveFile对象的映射，ActiveFile是活跃文件，它保存着处于写状态的数据块的一些附加信息。

数据节点和名字节点的交互

数据节点到名字节点的握手，注册，数据库上报和心跳;

名字节点保存并持久化了整个文件系统的文件目录树以及文件的数据块索引，但名字节点不持久化数据库的保存位置。HDFS启动时，数据节点会按期上报它上面保存的数据块信息。心跳上报过程当中，数据节点会发送可以描述当前节点负载的一些信息，如数据节点储存的容量，目前已使用的容量等，名字节点会根据这些信息估计数据节点的工做状态。

DataBlockScanner对数据块进行校验，

数据节点的启停

DataNode.main()是数据节点的入口方法，但只作一件事 sercureMain()方法，它经过createDataNode()建立并启动数据节点，而后经过dataNode对象的join()方法等待数据节点中止运行。

public static void main(String args[]) {

secureMain(args, null);

}

public static void secureMain(String[] args, SecureResources resources) {

try {

// ...todo something

DataNode datanode = createDataNode(args, null, resources);

if (datanode != null) {

datanode.join();

}

// do something....

} finally {

}

public static DataNode createDataNode(String args[], Configuration conf, SecureResources resources)

throws Exception {

DataNode dn = instanntiateDataNode(args, conf, resources);

runDatanodeDaemon(dn);

return dn;

}

名字节点的实现

hdfs的文件目录树，以及文件的数据块索引，即每一个文件对应的数据块列表

数据块和数据节点的对应关系。某个数据块保存在哪些数据节点的信息

文件系统

主要考虑文件和目录如何存储

linux中的i-node,索引节点，INode,INodeFile,INodeDirectory

命名空间镜像和编辑日志

文件目录由INode和其子类保存，若是节点掉电，数据将再也不存在，所以必须将信息保存到磁盘，镜像将保存某一时刻目录树的信息，它在HDFS中实现为FSImage,名字节点把命名空间镜像和编辑日志保存在current目录下，

远程调用过程

while(1){

msg=receive(anyClient)

unpack(msg,t1)

unoack(msg,t2)

unpack(msg,t3)

…

unpack(msg,tn)

func(t1,t2,…,tn)

pack(retMsg,a1)

pack(retMsg,an)

…

pack(retMsg,an)

send(theClient,retMsg)

}