hdfs的读写流程

hdfs写数据

1） 客户端通过Distributed?FileSystem??？橄騨amenode?请求上传文件，namenode?检查目标文件是否已存在，父目录是否存在。

2） namenode?返回是否可以上传。

3） 客户端请求第一个block?上传到哪几个datanode?服务器上。

4） namenode?返回3 个datanode?节点，分别为dn1、dn2、dn3。

5） 客户端通过FSDataOutputStream?模块请求dn1?上传数据，dn1?收到请求会继续调用dn2，然后dn2?调用dn3，将这个通信管道建立完成。

6） dn1、dn2、dn3 逐级应答客户端。

7） 客户端开始往dn1?上传第一个block（先从磁盘读取数据放到一个本地内存缓存），以

packet 为单位，dn1 收到一个packet 就会传给dn2，dn2 传给dn3；dn1 每传一个packet 会放入一个应答队列等待应答。

8） 当一个block?传输完成之后，客户端再次请求namenode?上传第二个block?的服务器。（重复执行3-7 步）。

hdfs写数据

1）客户端通过Distributed?FileSystem?向namenode?请求下载文件，namenode?通过查询元数据，找到文件块所在的datanode?地址。

2） 挑选一台datanode（就近原则，然后随机）服务器，请求读取数据。

3） datanode?开始传输数据给客户端（从磁盘里面读取数据输入流，以packet?为单位来做校验）。

4） 客户端以packet?为单位接收，先在本地缓存，然后写入目标文件。

namenode的工作机制

1） 第一阶段：namenode?启动

（1） 第一次启动namenode?格式化后，创建fsimage?和edits?文件。如果不是第一次启动，直接加载编辑日志和镜像文件到内存。

（2） 客户端对元数据进行增删改的请求。

（3）namenode?记录操作日志，更新滚动日志。

（4） namenode?在内存中对数据进行增删改查。

2） 第二阶段：Secondary?NameNode?工作

（1） Secondary?NameNode?询问namenode?是否需要checkpoint。直接带回namenode?是否检查结果。

（2）Secondary?NameNode 请求执行checkpoint。

（3） namenode?滚动正在写的edits?日志。

（4） 将滚动前的编辑日志和镜像文件拷贝到Secondary?NameNode。

（5）Secondary?NameNode 加载编辑日志和镜像文件到内存，并合并。

（6） 生成新的镜像文件fsimage.chkpoint。

（7） 拷贝fsimage.chkpoint 到namenode。

namenode?将fsimage.chkpoint 重新命名成fsimage。

datanode的工作机制

1） 一个数据块在datanode?上以文件形式存储在磁盘上，包括两个文件，一个是数据本身，一个是元数据包括数据块的长度，块数据的校验和，以及时间戳。

2） DataNode 启动后向namenode 注册，通过后，周期性（1?小时）的向namenode 上报所有的块信息。

3） 心跳是每3?秒一次，心跳返回结果带有namenode?给该datanode?的命令如复制块数据到另一台机器，或删除某个数据块。如果超过10?分钟没有收到某个datanode?的心跳，则认为该节点不可用。

集群运行中可以安全加入和退出一些机器

MapTask的工作机制

（1） Read?阶段：Map?Task?通过用户编写的RecordReader，从输入InputSplit 中解析出一个个key/value。

（2） Map 阶段：该节点主要是将解析出的key/value?交给用户编写map()函数处理，并产生一系列新的key/value。

（3）Collect?收集阶段：在用户编写map()函数中，当数据处理完成后，一般会调用

OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value 分区（调用

Partitioner），并写入一个环形内存缓冲区中。

（4）Spill?阶段：即“溢写”，当环形缓冲区满后，MapReduce?会将数据写到本地磁盘上，生成一个临时文件。需要注意的是，将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作。

溢写阶段详情：

步骤1：利用快速排序算法对缓存区内的数据进行排序，排序方式是，先按照分区编号

partition?进行排序，然后按照key?进行排序。这样，经过排序后，数据以分区为单位聚集在一起，且同一分区内所有数据按照key?有序。

步骤2：按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件output/spillN.out（N?表示当前溢写次数）中。如果用户设置了Combiner，则写入文件之前，对每个分区中的数据进行一次聚集操作。

步骤3：将分区数据的元信息写到内存索引数据结构SpillRecord 中，其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB，则将内存索引写到文件output/spillN.out.index 中。

（5）Combine?阶段：当所有数据处理完成后，MapTask?对所有临时文件进行一次合并，以确保最终只会生成一个数据文件。

当所有数据处理完后，MapTask 会将所有临时文件合并成一个大文件，并保存到文件

output/file.out 中，同时生成相应的索引文件output/file.out.index。

在进行文件合并过程中，MapTask 以分区为单位进行合并。对于某个分区，它将采用多轮递归合并的方式。每轮合并io.sort.factor（默认100）个文件，并将产生的文件重新加入待合并列表中，对文件排序后，重复以上过程，直到最终得到一个大文件。

让每个MapTask 最终只生成一个数据文件，可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。

Reduce Task的工作机制

（1） Copy?阶段：ReduceTask?从各个MapTask?上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。

（2） Merge?阶段：在远程拷贝数据的同时，ReduceTask?启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。

（3） Sort?阶段：按照MapReduce?语义，用户编写reduce()函数输入数据是按key?进行聚集的一组数据。为了将key?相同的数据聚在一起，Hadoop?采用了基于排序的策略。由于各个MapTask?已经实现对自己的处理结果进行了局部排序，因此，ReduceTask?只需对所有数据进行一次归并排序即可。

（4）Reduce?阶段：reduce()函数将计算结果写到HDFS?上。

Map Task 和Reduce Task 均会对数据（按照key）进行排序。该操作属于Hadoop 的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是否需要。默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。对于Map Task，它会将处理的结果暂时放到一个缓冲区中，当缓冲区使用率达到一定

阈值后，再对缓冲区中的数据进行一次排序，并将这些有序数据写到磁盘上，而当数据处理完毕后，它会对磁盘上所有文件进行一次合并，以将这些文件合并成一个大的有序文件。

对于Reduce Task，它从每个Map Task 上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则放到磁盘上，否则放到内存中。如果磁盘上文件数目达到一定阈值，则进行一次合并以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据写到磁盘上。当所有数据拷贝完毕后，Reduce Task 统一对内存和磁盘上的所有数据进行一次合并。

作业提交过程之yarn

（1） 作业提交

第0 步：client 调用job.waitForCompletion 方法，向整个集群提交MapReduce 作业。第1 步：client 向RM 申请一个作业id。

第2 步：RM 给client 返回该job 资源的提交路径和作业id。

第3 步：client 提交jar 包、切片信息和配置文件到指定的资源提交路径。第4 步：client 提交完资源后，向RM 申请运行MrAppMaster。

（2） 作业初始化

第5 步：当RM 收到client 的请求后，将该job 添加到容量调度器中。第6 步：某一个空闲的NM 领取到该job。

第7 步：该NM 创建Container，并产生MRAppmaster。第8 步：下载client 提交的资源到本地。

（3） 任务分配

第9 步：MrAppMaster 向RM 申请运行多个maptask 任务资源。

第10 步：RM 将运行maptask 任务分配给另外两个NodeManager，另两个NodeManager

分别领取任务并创建容器。

（4）任务运行

第11 步：MR 向两个接收到任务的NodeManager 发送程序启动脚本，这两个

NodeManager?分别启动maptask，maptask?对数据分区排序。

第?12?步：MrAppMaster?等待所有maptask?运行完毕后，向?RM?申请容器，运行?reduce?task。第13 步：reduce?task?向maptask?获取相应分区的数据。

第14 步：程序运行完毕后，MR 会向RM 申请注销自己。

（1） 进度和状态更新

YARN 中的任务将其进度和状态(包括counter)返回给应用管理器, 客户端每秒(通过

mapreduce.client.progressmonitor.pollinterval 设置)向应用管理器请求进度更新, 展示给用户。

（2） 作业完成

除了向应用管理器请求作业进度外, 客户端每5 分钟都会通过调用waitForCompletion() 来检查作业是否完成。时间间隔可以通过mapreduce.client.completion.pollinterval 来设置。作业完成之后, 应用管理器和container 会清理工作状态。作业的信息会被作业历史服务器存储以备之后用户核查。

数据倾斜问题的解决

方法1：抽样和范围分区

方法2：自定义分区

方法3：Combine

方法4：采用Map Join，尽量避免Reduce Join

小文件问题的解决

Hadoop Archive

Sequence?file

CombineFileInputFormat

开启JVM 重用