Hadoop - HDFS的数据流剖析
剖析文件的读取
为了了解客户端及与之交互的HDFS、namenode 和 datanode之间的数据流是什么样的，我们可以参考下图，该图显示了在读取文件时事件的发生顺序。
客户端通过调用FileSystem对象的open()方法来打开希望读取的文件，对于HDFS来说，这个对象是分布式文件系统（图中步骤1）的一个实例。DistributedFileSystem 通过使用RPC来调用namenode，以确定文件起始块的位置（步骤2）。对于每一个块，namenode返回存有该块副本的datanode地址。此外，这些datanode根据它们与客户端的距离来排序。如果该客户端本身就是一个datanode（比如，在一个MapReduce任务中），并保存有相应数据块的一个副本时，该节点就会从本地datanode读取数据。
DistributedFileSystem类返回一个FSDataInputStream对象（一个支持文件定位的数据流）给客户端并读取数据。FSDataInputStream类转而封装DFSInputStream对象，该对象管理着datanode和namenode的I/O。
接着，客户端对这个输入流调用 read()方法（步骤3）。存储着文件起始几个块的datanode地址的DFSInputStream随即连接距离最近的datanode。通过对数据反复调用read()方法，可以将数据从datanode传输到客户端（步骤4）。到达块的末端时，DFSInputStream关闭与该datanode的连接，然后寻找下一个块的最佳datanode（步骤5）。客户端只需要读取连续的流，并且对客户端都是透明的。
客户端从流中读取数据时，块是按照打开DFSInputStream与datanode新建连接的顺序读取的。它也会根据需要询问namenode来检索下一批数据块的datanode的位置。一旦客户端完成读取，就对FSDataInputStream调用close()方法（步骤6）。
在读取数据的时候，如果DFSInputStream在与datanode通信时遇到错误，会尝试从这个块的另一个最邻近datanode读取数据。它也会记住那个故障datanode，以保证以后不会反复读取该节点上后续的块。DFSInputStream也会通过校验和确认从datanode发来的数据是否完整。 如果发现有损坏的块，就在DFSInputStream试图从其他datanode读取其副本之前通知namenode。
这个设计的一个重点是，namenode告知客户端每个块中最佳的datanode，并让客户端直接连接到该datanode检索数据。由于数据流分散在急群众的所有datanode，所以这种设计能使HDFS可扩展到大量的并发客户端。同时，namenode只需要响应块位置的请求（这些信息存储在内存中，因而非常高效），无需响应数据请求，否则随着客户端数量的增长，namenode会很快成为瓶颈。
网络拓扑与Hadoop
在本地网络里，两个节点被称为"彼此近邻"是什么意思？在海量数据处理中，其主要限制因素是节点之间数据的传输速率--带宽很稀缺。这里的想法是将两个节点间的带宽作为距离的衡量标准。 不用衡量节点之间的带宽--实际上很难实现（它需要一个稳定的集群，并且在集群中两两节点对数量是节点数量的平方）--Hadoop为此采用一个简单的方法：把网络看做一棵树，两个节点间的距离是它们到最近共同祖先的距离总和。该树中的层次是没有预先设定的，但是相对于数据中心、几家和正在运行的节点，通常可以设定等级。具体想法是针对以下每个场景，可用带宽依次递减：
例如，假设有数据中心 d1 机架 r1 中的节点 n1.该节点可以表示为/d1/r1/n1。利用这种标记，这里给出四种距离描述：
最后，我们必须意识到Hadoop无法自行定义网络拓扑结构。它需要我们能够理解并辅助定义。 |
剖析文件的写入
接下来我们看看文件时如何写入HDFS的，尽管比较详细，但对于理解数据流还是很有用的，因为它清楚地说明了HDFS的一致模型。
我们要考虑的情况是如何新建一个文件，把数据写入该文件，最后关闭该文件。见下图。
客户端通过对DistributedFileSystem对象调用create()函数来新建文件（步骤1）。DistributedFileSystem对namenode创建一个RPC调用，在文件系统的命名空间中新建一个文件，此时该文件中还没有相应的数据块（步骤2）。namenode执行各种不同的检查以确保这个文件不存在以及客户端有新建文件的权限。如果这些检查均通过，namenode就会为创建新文件记录一条记录；否则，文件创建失败并向客户端抛出一个IOException异常。DistributedFileSystem向客户端返回一个FSDataOutputStream对象，由此客户端可以开始写入数据。就像读取事件一样，FSDataOutputStream封装一个DFSoutPutstream对象，该对象负责处理datanode和namenode之间的通信。
在客户端写入数据时（步骤3），DFSOutputStream将它分成一个个的数据包，并写入内部队列，称为"数据队列"（data queue）。DataStreamer处理数据队列，它的责任是根据datanode列表来要求namenode分配适合的新块来存储数据复本。这一组datanode构成一个管线--我们假设复本数为3，所以管线中有3个节点。DataStreamer将数据包流式传输到管线中第一个datanode，该datanode存储数据包并将它发送到管线中的第二个datanode。同样，第二个datanode存储该数据包并且发送给管线中的第三个（也是最后一个）datanote（步骤4）。
DFSOutputStream也维护着一个内部数据包队列来等待datanode的收到确认回执，称为"确认队列"（ack queue）。收到管道中所有datanode确认信息后，该数据包才会从确认队列删除（步骤5）。
如果数据在写入期间datanode发生故障，则执行以下操作（对写入数据的客户端是透明的）。首先关闭管线，确认把队列中的所有数据包都添加回数据队列的最前端，以确保故障节点下游的datanode不会漏掉任何一个数据包。为存储在另一正常datanode的当前数据结块指定一个新的标识，并将该标识传送给namenode，以便故障datanode在恢复后可以删除存储的部分数据块。从管线中删除故障数据节点并且把余下的数据块写入管线中另外两个正常的datanode。namenode注意到块复本量不足时，会在另一个节点上创建一个新的复本。后续的数据块继续正常接受处理。
在一个块被写入期间可能会有多个datanode同时发生故障，但非常少见。只要写入了dfs.replication.min的复本数（默认为1），写操作就会成功，并且这个块可以在集群中异步复制，直到达到其目标复本数（dfs.replication的默认值为3）。
客户端完成数据的写入后，对数据流调用close()方法（步骤6）。该操作将剩余的所有数据包写入datanode管线，并在联系到namenode且发送文件写入完成信号之前，等待确认（步骤7）。namenode已经知道文件由哪些块组成（通过Datastreamer请求分配数据块），所以它在返回成功之前只需要等待数据块进行最小量的复制。
复本怎么放
namenode如何选择在哪个datanode存储复本（replica）？这里需要对可靠性、写入带宽和读取带宽进行权衡。例如，把所有复本都存储在一个节点损失的写入带宽最小，因为复制管线都是在同一个节点上运行，但这并不提供真实的冗余（如果节点发生故障，那么该块中的数据会丢失）。同时，同一机架上服务器间的读取带宽是很高的。另一个极端，把复本放在不同的数据中心可以最大限度地提高冗余，但带宽的损耗非常大。即使在同一数据中心（到目前为止，所有Hadoop集群均运行在同一数据中心内），也有许多不同的数据布局策略。其实，在发布的Hadoop 0.17.0版中改变了数据布局策略来复制保持数据块在集群内分布相对均匀。在1.x之后的发行版本，可即时选择数据块的布局策略。 Hadoop的默认布局策略是在运行客户端的节点上放第一个复本（如果客户端运行在集群之外，就随机选择一个节点，不过系统会避免挑选那些存储太慢或太忙的节点）。第二个复本放在与第一个不同且随机另外选择的机架中节点上（离架）。第三个复本与第二个复本放在同一个机架上，且随机选择另一个节点。其他复本放在集群中随机选择的节点上，不过系统会尽量避免在同一个机架上放太多复本。 一旦选定复本的放置位置，就根据网络拓扑创建一个管线（Pipeline）。如果复本数为3，则有下图所示的管线。
总的来说，这一方法不仅提供很好的稳定性（数据块存储在两个机架中）并实现很好的负载均衡，包括写入带宽（写入操作只需要遍历一个交换机）、读取性能（可以从两个机架中选择读取）和集群中块的均匀分布（客户端只在本地机架上写入一个块）。 |
