HDFS pipeline写 -- datanode

时间:2022-09-01 10:56:08

站在DataNode的视角,看看pipeline写的流程,本文不分析客户端部分,从客户端写数据之前拿到了3个可写的block位置说起。

每个datanode会创建一个线程DataXceiverServer,接收上游过来的TCP连接,对于每个新建的TCP连接,都会创建一个叫做DataXceiver的线程处理这个连接. 这个线程不断的从TCP连接中读op,然后调用processOp(op)处理这个op,这里以write block 这个op为例.

对于datanode来说,write block操作由DataXceiver的writeBlock函数实现.

大体步骤如下:

  1. new 一个BlockReceiver对象,随后用于接收上游(client或者datanode)的block数据.

  2. 根据传进来的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.

  3. receiveBlock()结束后,关掉和上下游的连接.

清空ack queue的逻辑由专门处理下游ack包的PacketResponder线程处理,逻辑如下:

  1. 如果datanode是pipeline的中间node(通过PacketResponder的type属性来决定,LAST_IN_PIPELINE和HAS_DOWNSTREAM_IN_PIPELINE),
    那么从下游读一个PipelineAck,从ack中拿到seqno,然后从ack queue中get(不删除)第一个packet,拿出seqno,记作expected_seq_no,然后比较是否相等,如果不相等,说明写出错. 如果seqno相同,往下.

  2. 如果从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信息汇报上去.

  3. 给从下游接收的ack回复给上游。
  4. 将packet从ack queue的头部删除。

可以看出,一个block的写操作对于每个data node来说,由两个线程参与,一个是DataXceiver,用于接收上游的数据,一个是PacketResponder,用于处理下游回来的ack。还没有接收到下游的ack并且没有给上游回复ack的packet都存在在ack queue中。

参考资料

hadoop-hdfs-2.4.1.jar

HDFS pipeline写 -- datanode的更多相关文章

  1. HDFS pipeline写 -- 客户端

    上一篇说了datanode端如何处理pipeline写请求的,这里主要看DFSClient. 这里以append为例, write差不多. 创建一个pipeline用于append操作的流程: Fil ...

  2. HDFS dfsclient写文件过程 源码分析

    HDFS写入文件的重要概念 HDFS一个文件由多个block构成.HDFS在进行block读写的时候是以packet(默认每个packet为64K)为单位进行的.每一个packet由若干个chunk( ...

  3. HDFS的写数据过程分析

    HDFS的写数据过程分析 我们通过FileSystem类可以操控HDFS, 那我们就从这里开始分析写数据到HDFS的过程. 在我们向 HDFS 写文件的时候,调用的是 FileSystem.creat ...

  4. Linux启动kettle及linux和windows中kettle往hdfs中写数据(3)

    在xmanager中的xshell运行进入图形化界面 sh spoon.sh 新建一个job

  5. hadoop之 解析HDFS的写文件流程

    文件是如何写入HDFS的 ? 下面我们来先看看下面的“写”流程图:  假如我们有一个文件test.txt,想要把它放到Hadoop上,执行如下命令: 引用         # hadoop fs  - ...

  6. hdfs webhdfs 写文件(create file)

    # _*_ coding=utf-8 _*_ import sys import os hosts = {} cmd1 = ''' curl -i -X PUT "http://%s:500 ...

  7. HDFS 读/写数据流程

    1. HDFS 写数据流程 客户端通过 Distributed FileSystem 模块向 NameNode 请求上传文件, NameNode 检查目标文件是否已存在,父目录是否存在: NameNo ...

  8. HDFS数据节点DataNode未启动解决方法

    在解决这个问题的过程中,我又是积累了不少经验... 首先让我搞了很久的问题是,书上说进程全部启动的命令是/bin/start-all.sh,但是当我执行的时候显示command not found.后 ...

  9. HDFS 手写mapreduce单词计数框架

    一.数据处理类 package com.css.hdfs; import java.io.BufferedReader; import java.io.IOException; import java ...

随机推荐

  1. 无题 MVC

    1. MVC 里controller 返回匿名类型, 在View里是访问不了匿名类型的字段,因为它是Internal Private, 必须定义强类型 2. 扩展view的方法 public stat ...

  2. 给返回的injector增加了一个invoke方法

    我们知道,这里swap只需交换指针指向就可以,所以我们可以写出如下自定义版本的swap函数: 首先,它把一个独立的文件看成一个模块,比如上面的 hello.js 文件,就可以当成一个模块.模块的名称就 ...

  3. hdu 1175 连连看 (广搜,注意解题思维,简单)

    题目 解析见代码 #define _CRT_SECURE_NO_WARNINGS //这是非一般的最短路,所以广搜到的最短的路不一定是所要的路线 //所以应该把所有的路径都搜索出来,找到最短的转折数, ...

  4. 【原创】Kmeans算法 优缺点分析

    优点: 原理简单(靠近中心点),实现容易(1.2 天),聚类效果中上(依赖K的选择) 缺点: 1. 无法确定K的个数 (根据什么指标确定K) 2. 对离群点敏感 (容易导致中心点偏移) 3. 算法复杂 ...

  5. 转:Internal Sales Order (ISO) Process Flow

    本文介绍下内部销售订单Internal Sales Order(ISO)在Oracle EBS中的流程,内部销售订单和组织间转移(Inter-Organization Transfer,IOT)的作用 ...

  6. 【实习记】2014-08-18使用curl排错http头的content-length

        总结一,用curl排错Content-Length设置错误,误导了客户端. 访问/cgi-bin/txproj_list时,firebug显示总是不多不少15秒,调试其他问题时郁闷. fire ...

  7. php中的short_open_tag的作用

    在php的配置文件(php.ini)中有一个short_open_tag的值,开启以后可以使用PHP的段标签:(<? ?>). 同时,只有开启这个才可以使用 <?= 以代替 < ...

  8. android代码实现关机

      1.API没有开放,需要提升为syetem app级别! 2.android 模块编译,mm 命令 2.1.先进入顶层  source build/envsetup.sh 2.2.进入目录   m ...

  9. CheckBox和控件Foreach循环

    .aspx代码例如以下: <%@ Page Language="C#" AutoEventWireup="true" CodeFile="Sam ...

  10. JS数组添加删除

    栈是一种LIFO(Last-In-First-Out,后进先出)的数据结构著作权归作者所有.商业转载请联系作者获得授权,非商业转载请注明出处.原文: https://www.w3cplus.com/j ...