站在DataNode的视角,看看pipeline写的流程,本文不分析客户端部分,从客户端写数据之前拿到了3个可写的block位置说起。
每个datanode会创建一个线程DataXceiverServer,接收上游过来的TCP连接,对于每个新建的TCP连接,都会创建一个叫做DataXceiver的线程处理这个连接. 这个线程不断的从TCP连接中读op,然后调用processOp(op)处理这个op,这里以write block 这个op为例.
对于datanode来说,write block操作由DataXceiver的writeBlock函数实现.
大体步骤如下:
new 一个BlockReceiver对象,随后用于接收上游(client或者datanode)的block数据.
-
根据传进来的DatanodeInfo数组,向数组的第一个元素代表的datanode建立TCP连接,targets参数是从上游的TCP连接中解析出来的,逻辑在Receiver的opWriteBlock方法中,Receiver是DataXceiver的基类.然后调用Sender的writeBlock方法给下游datanode发送write block相关元信息,包括DatanodeInfo数组(刨去第一个元素),clientname,block的当前gs,minBytesRcvd,maxBytesRcvd(对于append,recovery操作有用)等。然后读取下游的回复封装在BlockOpResponseProto对象中,可以通过内部成员firstBadLink知道建pipeline中第一个失败的datanode节点。接着将BlockOpResponseProto回复给上游
(datanode或者client),最后调用第一步new的BlockReceiver的receiveBlock方法用于接收一个完整的block.如下:receiveBlock内部根据clientname发现是一个客户端在写block,创建一个PacketResponder线程用于处理下游datanode对packet的ack.PacketResponder后面分析。接着,不断的调用receivePacket()方法从上游(datanode或者client)接收一个个的packet,接收一个完整的packet的逻辑是由内部的PacketReceiver来处理的.
对于一个接收到的packet,写入block file文件,同时checksum信息写meta文件,然后放入PacketResponder的ack queue队列,然后将packet写给下游的datanode。最后调用PacketResponder的 close方法,这个方法会等到ack queue为空,即所有packet都已经从下游收到,并且已经给上游ack. receiveBlock()结束后,关掉和上下游的连接.
清空ack queue的逻辑由专门处理下游ack包的PacketResponder线程处理,逻辑如下:
如果datanode是pipeline的中间node(通过PacketResponder的type属性来决定,LAST_IN_PIPELINE和HAS_DOWNSTREAM_IN_PIPELINE),
那么从下游读一个PipelineAck,从ack中拿到seqno,然后从ack queue中get(不删除)第一个packet,拿出seqno,记作expected_seq_no,然后比较是否相等,如果不相等,说明写出错. 如果seqno相同,往下.-
如果从ack queue中get的packet是block的最后一个packet,说明一个block接收完成.那么调用finalizeBlock方法.finalizeBlock方法逻辑如下:
关闭block file和meta file文件,调用FsDatasetImpl的finalizeBlock(block)将block文件以及对应的meta文件移动到对应的block pool下的finalized目录下,然后生成一个FinalizedReplica对象,将bpid->FinalizedReplica的映射关系记录在内存中的volumnMap中,对象位于FsDatasetImpl下的ReplicaMap volumnMap(从ReplicaMap中定位一个ReplicaInfo,需要拿着bpid和block id去找)最后调用datanode的closeBlock()方法,将block回报给namenode,该方法逻辑如下:
拿着block的bpid从BlockPoolManager中拿到相应的BPOfferService,通知namenode这个block。在data node这边,data node和每个namenode的接口由一
个BPServiceActor来承担,这是一个线程, 这个线程会向namenode汇报received block或者指示namenode去删除block.最后调用DatanodeProtocolClientSideTranslatorPB bpNamenode的blockReceivedAndDeleted()将block信息汇报上去. - 给从下游接收的ack回复给上游。
将packet从ack queue的头部删除。
可以看出,一个block的写操作对于每个data node来说,由两个线程参与,一个是DataXceiver,用于接收上游的数据,一个是PacketResponder,用于处理下游回来的ack。还没有接收到下游的ack并且没有给上游回复ack的packet都存在在ack queue中。
参考资料
hadoop-hdfs-2.4.1.jar
HDFS pipeline写 -- datanode的更多相关文章
-
HDFS pipeline写 -- 客户端
上一篇说了datanode端如何处理pipeline写请求的,这里主要看DFSClient. 这里以append为例, write差不多. 创建一个pipeline用于append操作的流程: Fil ...
-
HDFS dfsclient写文件过程 源码分析
HDFS写入文件的重要概念 HDFS一个文件由多个block构成.HDFS在进行block读写的时候是以packet(默认每个packet为64K)为单位进行的.每一个packet由若干个chunk( ...
-
HDFS的写数据过程分析
HDFS的写数据过程分析 我们通过FileSystem类可以操控HDFS, 那我们就从这里开始分析写数据到HDFS的过程. 在我们向 HDFS 写文件的时候,调用的是 FileSystem.creat ...
-
Linux启动kettle及linux和windows中kettle往hdfs中写数据(3)
在xmanager中的xshell运行进入图形化界面 sh spoon.sh 新建一个job
-
hadoop之 解析HDFS的写文件流程
文件是如何写入HDFS的 ? 下面我们来先看看下面的“写”流程图: 假如我们有一个文件test.txt,想要把它放到Hadoop上,执行如下命令: 引用 # hadoop fs - ...
-
hdfs webhdfs 写文件(create file)
# _*_ coding=utf-8 _*_ import sys import os hosts = {} cmd1 = ''' curl -i -X PUT "http://%s:500 ...
-
HDFS 读/写数据流程
1. HDFS 写数据流程 客户端通过 Distributed FileSystem 模块向 NameNode 请求上传文件, NameNode 检查目标文件是否已存在,父目录是否存在: NameNo ...
-
HDFS数据节点DataNode未启动解决方法
在解决这个问题的过程中,我又是积累了不少经验... 首先让我搞了很久的问题是,书上说进程全部启动的命令是/bin/start-all.sh,但是当我执行的时候显示command not found.后 ...
-
HDFS 手写mapreduce单词计数框架
一.数据处理类 package com.css.hdfs; import java.io.BufferedReader; import java.io.IOException; import java ...
随机推荐
-
无题 MVC
1. MVC 里controller 返回匿名类型, 在View里是访问不了匿名类型的字段,因为它是Internal Private, 必须定义强类型 2. 扩展view的方法 public stat ...
-
给返回的injector增加了一个invoke方法
我们知道,这里swap只需交换指针指向就可以,所以我们可以写出如下自定义版本的swap函数: 首先,它把一个独立的文件看成一个模块,比如上面的 hello.js 文件,就可以当成一个模块.模块的名称就 ...
-
hdu 1175 连连看 (广搜,注意解题思维,简单)
题目 解析见代码 #define _CRT_SECURE_NO_WARNINGS //这是非一般的最短路,所以广搜到的最短的路不一定是所要的路线 //所以应该把所有的路径都搜索出来,找到最短的转折数, ...
-
【原创】Kmeans算法 优缺点分析
优点: 原理简单(靠近中心点),实现容易(1.2 天),聚类效果中上(依赖K的选择) 缺点: 1. 无法确定K的个数 (根据什么指标确定K) 2. 对离群点敏感 (容易导致中心点偏移) 3. 算法复杂 ...
-
转:Internal Sales Order (ISO) Process Flow
本文介绍下内部销售订单Internal Sales Order(ISO)在Oracle EBS中的流程,内部销售订单和组织间转移(Inter-Organization Transfer,IOT)的作用 ...
-
【实习记】2014-08-18使用curl排错http头的content-length
总结一,用curl排错Content-Length设置错误,误导了客户端. 访问/cgi-bin/txproj_list时,firebug显示总是不多不少15秒,调试其他问题时郁闷. fire ...
-
php中的short_open_tag的作用
在php的配置文件(php.ini)中有一个short_open_tag的值,开启以后可以使用PHP的段标签:(<? ?>). 同时,只有开启这个才可以使用 <?= 以代替 < ...
-
android代码实现关机
1.API没有开放,需要提升为syetem app级别! 2.android 模块编译,mm 命令 2.1.先进入顶层 source build/envsetup.sh 2.2.进入目录 m ...
-
CheckBox和控件Foreach循环
.aspx代码例如以下: <%@ Page Language="C#" AutoEventWireup="true" CodeFile="Sam ...
-
JS数组添加删除
栈是一种LIFO(Last-In-First-Out,后进先出)的数据结构著作权归作者所有.商业转载请联系作者获得授权,非商业转载请注明出处.原文: https://www.w3cplus.com/j ...