什么是hadoop？

hadoop 是一个可编程和运行分布式应用，用来处理大数据的开源框架。

Hadoop主要子项目　　

Hadoop Common: 在0.20及以前的版本中，包含HDFS、MapReduce和其他项目公共内容，从0.21开始HDFS和MapReduce被分离为独立的子项目，其余内容为Hadoop Common

HDFS: Hadoop 分布式文件系统 (Distributed File System) － HDFS (Hadoop Distributed File System)

MapReduce：并行计算框架，0.20前使用 org.apache.hadoop.mapred 旧接口，0.20版本开始引入org.apache.hadoop.mapreduce的新API

HBase: 类似Google BigTable的分布式NoSQL列数据库。（HBase 和 Avro 已经于2010年5月成为顶级 Apache 项目）

Hive：数据仓库工具，由Facebook贡献。

Zookeeper：分布式锁设施，提供类似Google Chubby的功能，由Facebook贡献。

Avro：新的数据序列化格式与传输工具，将逐步取代Hadoop原有的IPC机制。

和其他分布式架构比较：

SETI@home(就是寻找外星人那个)：是将数据传送到要计算的地方（运行屏保的计算机），经过计算，再将计算结果传回到数据中心。

hadoop的做法：将代码向数据所在的地方迁移。

和普通的数据库比较：

1.处理的对象不一样。

传统数据库处理的是 结构化的数据,如 表的结构都是固定的。结构化查询语言(Structured Query Language)简称SQL.

hadoop更多的是处理半结构化的数据或者非结构化的数据，如分析日志记录，统计字符出现的次数等等。

2.拓展方式不一样

hadoop的拓展是向外拓展，即需要扩容的时候，增加普通的机器。

普通数据库拓展是向高拓展，即更换更好的机器。（当然也可以向外拓展，这点有待讨论...）

3.hadoop用键值对代替数据表

传统数据库是让数据以某种模式存放在具有关系数据库的模式中。基于这种模式来对数据进行处理。hadoop将数据转化为键/值对来进行处理

4.hadoop用函数式编程（MapReduce）代替sql

5.hadoop用离线批量处理代替在线处理（hadoop为离线处理和大规模数据分析而设计，

更适合于 一次写入，多次读取 的情况，类似于sql的数据仓库）

简单的一致性模型

HDFS应用需要一个“一次写入多次读取”的文件访问模型。一个文件经过创建、写入和关闭之后就不需要改变。这一假设简化了数据一致性问题，并且使高吞吐量的数据访问成为可能。Map/Reduce应用或者网络爬虫应用都非常适合这个模型。目前还有计划在将来扩充这个模型，使之支持文件的附加写操作。

Hadoop Distributed File System，简称HDFS，是一个分布式文件系统。

1. 特性：

1.1. 大数据集

运行在HDFS上的应用具有很大的数据集。可以是从GB到TB级的。

1.2. 高容错性

在hadoop集群环境下，每份数据都被保存在多个结点里边。一个结点的不可用不会导致该结点存储的数据不可用。

1.3. 高吞吐量

跑在Hdoop上的应用，更多的是做批处理，HDFS的设计中更多的考虑到了数据批处理，而不是用户交互处理。比之数据访问的低延迟问题，更关键的在于数据访问的高吞吐量。

1.4. 简单一致的模型

采用了"一次写入，多次读取"的模型。这一假设简化了数据一致性问题，并且使高吞吐量的数据访问成为可能。Map/Reduce应用或者网络爬虫应用都非常适合这个模型。目前还有计划在将来扩充这个模型，使之支持文件的附加写操作。

1.5. 可移植性

HDFS在设计的时候就考虑到平台的可移植性。这种特性方便了HDFS作为大规模数据应用平台的推广。 （可以这样理解：基于java，并且在HDFS是在本地FileSystem之上抽象出来的文件系统）

1.6. 移动计算比移动数据更高效

一个应用请求的计算，离它操作的数据越近就越高效，在数据达到海量级别的时候更是如此。因为这样就能降低网络阻塞的影响，提高系统数据的吞吐量。将计算移动到数据附近，比之将数据移动到应用所在显然更好。HDFS为应用提供了将它们自己移动到数据附近的接口。

一个简单的例子：程序X在计算机A上，数据在计算机A,B和C上。普通的做法是在A上运行程序X，读取A，B，C上的数据，然后进行计算（或者是边读取边计算）。hadoop的做法是：将程序X同时也发送到B和C上，A，B，C同时读取本地计算机上的数据进行计算。得出结果之后再进行汇总（当然这里只是通俗的举了个例子，实际上hadoop处理数据更为复杂）。这样就避免了在网络上传输大量数据导致的等待和B，C计算能力的浪费。

2. 概念：

nameNode 和 dataNode

HDFS采用了master/slave架构。HDFS由一个nameNode和若干个dataNode组成。

2.1 nameNode

nameNode类似于一个仓库管理员，负责记录数据存储在哪些仓库（dataNode）的哪些地方，同时也负责和客户端（Client）对文件的访问。nameNode作为一个中心服务器。因此当nameNode出现故障时是比较致命的。为了防止出现恶劣的后果，通常集群环境下还有一个Secondary NameNode 作为备份。

2.2 dataNode

dataNode负责文件的存储（可以理解为仓库）。

一个文件被分为多个文件块存储在一组dataNode里边。Namenode执行文件系统的名字空间操作，比如打开、关闭、重命名文件或目录。它也负责确定数据块到具体Datanode节点的映射。Datanode负责处理文件系统客户端的读写请求。在Namenode的统一调度下进行数据块的创建、删除和复制。

3. 文件流

3.1 读文件

客户端(client)用FileSystem的open()函数打开文件，DistributedFileSystem用RPC调用元数据节点，得到文件的数据块信息。对于每一个数据块，元数据节点返回保存数据块的数据节点的地址。DistributedFileSystem返回FSDataInputStream给客户端，用来读取数据。客户端调用stream的read()函数开始读取数据。DFSInputStream连接保存此文件第一个数据块的最近的数据节点。Data从数据节点读到客户端(client)，当此数据块读取完毕时，DFSInputStream关闭和此数据节点的连接，然后连接此文件下一个数据块的最近的数据节点。当客户端读取完毕数据的时候，调用FSDataInputStream的close函数。

整个过程就是如图所示

3.2 写文件

客户端调用create()来创建文件，DistributedFileSystem用RPC调用元数据节点，在文件系统的命名空间中创建一个新的文件。元数据节点首先确定文件原来不存在，并且客户端有创建文件的权限，然后创建新文件。DistributedFileSystem返回DFSOutputStream，客户端用于写数据。客户端开始写入数据，DFSOutputStream将数据分成块，写入data queue。Data queue由Data Streamer读取，并通知元数据节点分配数据节点，用来存储数据块(每块默认复制3块)。分配的数据节点放在一个pipeline里。Data Streamer将数据块写入pipeline中的第一个数据节点。第一个数据节点将数据块发送给第二个数据节点。第二个数据节点将数据发送给第三个数据节点。DFSOutputStream为发出去的数据块保存了ack queue，等待pipeline中的数据节点告知数据已经写入成功。如果数据节点在写入的过程中失败：关闭pipeline，将ack queue中的数据块放入data queue的开始。

整个过程如图所示：

4. 使用命令行操作HDFS

1.添加目录 hadoop fs -mkdir /abc

就在HDFS下创建了一个abc的目录

2.列出HDFS根目录下的文件 hadoop fs -ls /

列出根目录下的所有文件 hadoop fs -ls

3.复制文件 hadoop fs -put /tianlong/test /abc

4.查看文件 hadoop fs -cat /abc/test

5.从HDFS上取文件 hadoop fs -get /abc/test ~/test.txt 把HDFS里的test文件 取回 当前用户目录的test.txt

5. 编程操作HDFS

下边简单介绍几个常用的API，具体的例子网上有很多，就不重复贴了。

FileSystem Hadoop文件系统通过Hadoop Path 对象来代表文件，是一个通用的文件系统API。

　Hadoop中关于文件操作类基本上全部是在"org.apache.hadoop.fs"包中，这些API能够支持的操作包含：打开文件，读写文件，删除文件等。

Hadoop类库中最终面向用户提供的接口类是FileSystem，该类是个抽象类，只能通过来类的get方法得到具体类。get方法存在几个重载版本，常用的是这个：

static FileSystem get(Configuration conf) throws IOException;

static FileSystem get(URI uri, Configuration conf) throws IOException;

或者从Path对象中获得FileSystem实例

Path dst = new Path("hdfs://192.168.0.111:9000/home");

// 得到dst的FileSystem.

FileSystem hdfs = dst.getFileSystem(conf);

　　该类封装了几乎所有的文件操作，例如mkdir，delete等。综上基本上可以得出操作文件的程序库框架：

operator()

{

得到Configuration对象

得到FileSystem对象

进行文件操作

}

MapReduce的作业（job）是由客户端提交给Hadoop集群的。一个job包括了输入数据，MapReduce程序和配置信息。Hadoop将作业非为若干个任务（task）来执行，其中包括map任务和reduce任务。

有两类节点和task的执行相关，一类是jobtracker,另外一类是tasktracker。

jobtracker通过调度tasktracker上运行的任务来协调所有运行在系统上的作业。

tasktracker在运行任务的同时将运行的进度返回给jobtracker，jobtracker由此记录每项作业的情况，如果有tasktracker执行任务失败，则jobtracker可以在另外一个tasktracker上重新调度失败的任务。

Hadoop将MapReduce的输入数据划分成等长的数据块，被称为输入分片。hadoop为每个分片构建一个map任务，并该任务运行由用户定义的map函数对分片中的每条记录。

对于大部分作业来说，一个合理的分片大小趋向于HDFS的一个块大小。

Hadoop在存储有输入数据的节点上运行map任务，可以获得最佳性能（数据本地化优化）。

这就是最佳分片大小与HDFS块大小相同的原因：因为他是确保可以存储在单个结点上的最大输入块大小。如果分片夸两个数据库，那么对任何一个HDFS节点，基本上都不可能同时存储这两个数据库，因此分片中的部分数据需要通过网络传输到Map任务节点。与使用本地数据运行整个map相比，就显得笨拙了。

map-reduce的运行流程：

1.map-reduce 程序启动作业

2.jobclient向jobTracker请求一个新的jobId。同事检查作业输入和输出说明。如果不合法就返回异常

3.jobclient 将运行作业所需要的资源复制到一个以作业ID命名的目录下jobtracker的文件系统中。作业jar的副本较多，因此在运行作业的任务时，集群中有很多副本可供taskTracker访问。

4.jobclient告知jobTracker作业准备执行

5.jboTracker接收到对其submitJob()方法的调用后，会把此调用放入一个内部队列中，交由作业调度进行调度，并对其初始化。

6.为了创建任务运行列表，作业调度器首先从共享文件系统中获取jobClient以计算好的输入分片信息，然后为每一个分片创建map任务。创建reduce任务数量由jobConf的mapred.reduce.task属性决定.然后调度器创建相应数量的要运行的reduce任务。任务在此时被指定ID

7.tasktracker运行一个简单的循环来定期发送“心跳”给jobtracker，用来告诉jobtracker它自己是否还存活和是否准备好运行新的任务。如果是，jobtracker会为他分配一个任务，并使用“心跳”返回值与tasktracker进行通信。

在为tasktracker选择任务(task)之前，jobtracker首先要选定任务所在的作业（job）。选择好作业之后jobtracker就可以作为该作业选定一个任务。对于map任务和reduce任务，tasktacker有固定数量的任务槽。eg:一个tasktracker可能同时运行两个map任务和reduce任务。默认调度器在处理reduce任务之前会填满空闲的map任务草。因此，如果tasktracker至少有一个空闲的map任务槽，jobtracker会为它选择一个map任务，否则选择一个reduce任务。

在选择reduce任务的时候，jobtracker简单的从待运行的reduce任务列表中选取下一个来执行，用不着考虑数据的本地化。对于一个map任务jobtracker会考虑tasktracker的网络位置，并选取一个距离与其输入分片最近的tasktracker。最理想的情况是 数据本地化的（任务运行在和输入分片在同一个机器上）。次之是 机架本地化的。

tasktracker执行任务：通过共享文件系统把作业的jar文件复制到tasktracker所在的文件系统，从而实现作业的jar文件本地化，同时，tasktracker将应用程序所需要的全部文件从分布式缓存复制到本地磁盘。然后tasktracker为任务新建一个本地工作目录，并把jar文件中的内容解压到这个文件夹下，然后tasktracker新建一个TaskRunner实例来运行该任务。TaskRunner启动一个新的JVM来运行每个任务，以便用户自定义的map和reduce函数的人和问题不会影响到tasktracker。子进程通过接口与父进程通信。任务的资进程每隔几秒便告知父进程它的进度，直到任务完成。

当jobtracker收到作业的最后一个任务已完成的通知后，便把作业状态设置为“成功”。然后在jobclient查询状态时直到任务已经完成。于是jobclient打印一条消息告知用户，然后从runJob方法返回。

最后jobtracker清空作业的工作状态，指示tasktracker也清空作业的工作状态（eg:删除中间输出）。

map任务将结果写入本地硬盘，而非HDFS。因为map任务的结果是中间结果，要给reduce任务进行再次处理，处理完之后map任务的结果就没有价值了，通常是被删掉。HDFS上的同一份数据，通常情况下是要备份的。如果存入HDFS，那么就有些小题大做了。

MapReduce确保每个reducer的输入都按键排序。系统执行排序的过程---将map输出作为输入传给reducer，称为为shuffle。

map产生结果之后会先在内存中，过了一定的量之后开始写入本地硬盘。

写入磁盘之前，线程先根据数据最终要送到的reducer把数据划分成相应的分区（partition）。在每个分区中，后台线程按键进行行内排序，如果有一个combiner，它会在排序后的输入上运行。

一旦缓冲区达到阈值，就开始新建溢出文件。所以在map任务写完其最后一个输出记录之后，会有几个溢出写文件。在溢出写文件被合并成一个已经分区且排序出的输出文件。

combiner的意义在于让写入本地磁盘和传到reducer上的数据更加紧凑。

reducer通过http方式得到输出文件的分区。map输入文件位于运行map任务的tasktracker的本地磁盘上。现在，tasktracker需要为分区文件运行reduce任务。这个reduce任务通常需要集群上若干个map任务的输出作为其特殊的分区文件。每个map任务完成时间可能不同。因此只要有一个任务完成，reduce任务就开始复制其输出。如果map输出相当小，则会被复制到 reduce tasktracker的内存中，否则会被复制到磁盘。一旦内存缓冲区达到阀值大小或者达到map输出阀值，则合并后溢出写到磁盘中。随着磁盘上的副本增多，后台线程会将他们合并为更大，排好序的文件。为后边的合并节省一些时间。复制完所有map输出被复制期间，reduce任务进入合并阶段。这阶段将合并map输出，维持其顺序排序。

在reduce阶段，对已排序输出中的每个键都调用reduce函数。此阶段的输出直接写到HDFS中。