通過Thrift訪問HDFS分布式文件系統的性能瓶頸分析

引言

　　Hadoop提供的HDFS布式文件存儲系統，提供了基於thrift的客戶端訪問支持，但是因為Thrift自身的訪問特點，在高並發的訪問情況下，thrift自身結構可能將會成為HDFS文件存儲系統的一個性能瓶頸。我們先來看一下一不使用Thrfit方式訪問HDFS文件系統的業務流程。

一、HDFS文件讀取流程

　　

流程說明：

1. 使用HDFS提供的客戶端開發庫Client，向遠程的Namenode發起RPC請求；
2. Namenode會視情況返回文件的部分或者全部block列表，對於每個block，Namenode都會返回有該block拷貝的DataNode地址；
3. 客戶端開發庫Client會選取離客戶端最接近的DataNode來讀取block；如果客戶端本身就是DataNode,那么將從本地直接獲取數據.
4. 讀取完當前block的數據后，關閉與當前的DataNode連接，並為讀取下一個block尋找最佳的DataNode；
5. 當讀完列表的block后，且文件讀取還沒有結束，客戶端開發庫會繼續向Namenode獲取下一批的block列表。
6. 讀取完一個block都會進行checksum驗證，如果讀取datanode時出現錯誤，客戶端會通知Namenode，然后再從下一個擁有該block拷貝的datanode繼續讀。

二、HDFS文件寫入流程

流程說明：

1. 使用HDFS提供的客戶端開發庫Client，向遠程的Namenode發起RPC請求；
2. Namenode會檢查要創建的文件是否已經存在，創建者是否有權限進行操作，成功則會為文件創建一個記錄，否則會讓客戶端拋出異常；
3. 當 客戶端開始寫入文件的時候，開發庫會將文件切分成多個packets，並在內部以數據隊列"data queue"的形式管理這些packets，並向Namenode申請新的blocks，獲取用來存儲replicas的合適的datanodes列表， 列表的大小根據在Namenode中對replication的設置而定。
4. 開始以pipeline（管道）的形式將packet寫入所 有的replicas中。開發庫把packet以流的方式寫入第一個datanode，該datanode把該packet存儲之后，再將其傳遞給在此 pipeline中的下一個datanode，直到最后一個datanode，這種寫數據的方式呈流水線的形式。
5. 最后一個datanode成功存儲之后會返回一個ack packet，在pipeline里傳遞至客戶端，在客戶端的開發庫內部維護着"ack queue"，成功收到datanode返回的ack packet后會從"ack queue"移除相應的packet。
6. 如 果傳輸過程中，有某個datanode出現了故障，那么當前的pipeline會被關閉，出現故障的datanode會從當前的pipeline中移除， 剩余的block會繼續剩下的datanode中繼續以pipeline的形式傳輸，同時Namenode會分配一個新的datanode，保持 replicas設定的數量。

三、關鍵詞

　　HDFSClient通過文件IO操作最終實現是通過直接訪問DataNode進行。

四、Thrift的訪問流程：猜測版

　　

流程說明：

1.ThriftClient客戶端將操作命令傳給ThriftServer。

2.ThriftServer調用HDFSClient接口API實現HDFS讀寫操作，操作流程如二和三所示。

五、疑問

　　與DataNode發生數據交換的到底是ThriftServer還是ThriftClient，如果是ThriftServer，那么多個ThriftClient並行訪問時，ThriftServer必將成為HDFS訪問的性能瓶頸；如果是ThriftClient直接訪問DataNode，那么理論依據何在呢？

六、示例程序

　　下面是一個基於Thrift實現的HDFS客戶端程序，實現了文件的訪問和創建和讀取

// HdfsDemo.cpp : Defines the entry point for the console application.
//

#include "stdafx.h"
#include <iostream>
#include <string>
#include <boost/lexical_cast.hpp>
#include <protocol/TBinaryProtocol.h>
#include <transport/TSocket.h>
#include <transport/TTransportUtils.h>
#include "ThriftHadoopFileSystem.h"

#ifndef _WIN32_WINNT
#define _WIN32_WINNT 0x0500
#endif 
using namespace std;
using namespace apache::thrift;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;

int _tmain(int argc, _TCHAR* argv[])
{
    if (argc < 3) 
    {
        std::cerr << "Invalid arguments!\n" << "Usage: DemoClient host port" << std::endl;
        //return -1;
    }
    boost::shared_ptr<TTransport> socket(new TSocket("192.168.230.133", 55952));//boost::lexical_cast<int>(argv[2])));
    boost::shared_ptr<TTransport> transport(new TBufferedTransport(socket));
    boost::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
    ThriftHadoopFileSystemClient client(protocol);
    try 
    {
        transport->open();
        Pathname path;
        //01_create directory
        path.__set_pathname("/user/hadoop");
        if(client.exists(path) == true)
        {
            printf("path is exists.\r\n");
        }
        else
        {
            printf("path is not exists.");
            //return 0;
        }
        //02_put file
        Pathname filepath;
        filepath.__set_pathname("/user/hadoop/in/test1.txt");
        /*
        FILE* localfile = fopen("E:\\project\\Hadoop\\HdfsDemo\\Debug\\hello.txt","rb");
        if (localfile == NULL)
        {
            transport->close();
            return 0;
        }
 ThriftHandle hdl;  58  client.create(hdl,filepath);  59         while (true)
        {
            char data[1024];
            memset(data,0x00,sizeof(data));
            size_t Num = fread(data,1,1024,localfile);
            if (Num <= 0)
            {
                break;
            }
 client.write(hdl,data);  69         }
        fclose(localfile);
 client.close(hdl);  72         */
        //03_get file
        /*
        ThriftHandle hd2;
        FileStatus stat1;
        client.open(hd2,filepath);
        client.stat(stat1,filepath);
        int index = 0;
        while(true)
        {
            string data;
            if (stat1.length <= index)
            {
                break;
            }
            client.read(data,hd2,index,1024);

            index += data.length();
            printf("==%s\r\n",data.c_str());
        }
        client.close(hd2);
        */

        //04_list files
        std::vector<FileStatus> vFileStatus;
        client.listStatus(vFileStatus,path);
        for (int i=0;i<vFileStatus.size();i++)
        {
            printf("i=%d file=%s\r\n",i,vFileStatus[i].path.c_str());
        }
        transport->close();
    } catch (const TException &tx) {
    std::cerr << "ERROR: " << tx.what() << std::endl;
    }
    getchar();
    return 0;
}

 七、源碼分析

　　1.文件創建：

    /**
      * Create a file and open it for writing, delete file if it exists
      */
    public ThriftHandle createFile(Pathname path, 
                                   short mode,
                                   boolean  overwrite,
                                   int bufferSize,
                                   short replication,
                                   long blockSize) throws ThriftIOException {
      try {
        now = now();
        HadoopThriftHandler.LOG.debug("create: " + path +
                                     " permission: " + mode +
                                     " overwrite: " + overwrite +
                                     " bufferSize: " + bufferSize +
                                     " replication: " + replication +
                                     " blockSize: " + blockSize);
        FSDataOutputStream out = fs.create(new Path(path.pathname), 
                                           new FsPermission(mode),
                                           overwrite,
                                           bufferSize,
                                           replication,
                                           blockSize,
                                           null); // progress
        long id = insert(out);
        ThriftHandle obj = new ThriftHandle(id);
        HadoopThriftHandler.LOG.debug("created: " + path + " id: " + id);
        return obj;
      } catch (IOException e) {
        throw new ThriftIOException(e.getMessage());
      }
    }

　　ThriftHandle的兩端到底是誰呢？是ThriftClient和DataNode？還是ThriftServer與DataNode？

　　2.文件寫入

    public boolean write(ThriftHandle tout, String data) throws ThriftIOException {
      try {
        now = now();
        HadoopThriftHandler.LOG.debug("write: " + tout.id);
        FSDataOutputStream out = (FSDataOutputStream)lookup(tout.id);
        byte[] tmp = data.getBytes("UTF-8");
        out.write(tmp, 0, tmp.length);
        HadoopThriftHandler.LOG.debug("wrote: " + tout.id);
        return true;
      } catch (IOException e) {
        throw new ThriftIOException(e.getMessage());
      }
    }

　　寫入時依賴的還是ThriftHandle？　　

　　3.文件讀取

    /**
     * read from a file
     */
    public String read(ThriftHandle tout, long offset,
                       int length) throws ThriftIOException {
      try {
        now = now();
        HadoopThriftHandler.LOG.debug("read: " + tout.id +
                                     " offset: " + offset +
                                     " length: " + length);
        FSDataInputStream in = (FSDataInputStream)lookup(tout.id);
        if (in.getPos() != offset) {
          in.seek(offset);
        }
        byte[] tmp = new byte[length];
        int numbytes = in.read(offset, tmp, 0, length);
        HadoopThriftHandler.LOG.debug("read done: " + tout.id);
        return new String(tmp, 0, numbytes, "UTF-8");
      } catch (IOException e) {
        throw new ThriftIOException(e.getMessage());
      }
    }

 八、遺留問題

　　ThriftHandle可以看做是Socket連接句柄，但是他的兩端到底是誰呢？如果是ThriftClient代表的客戶端則一切OK，那么我該如何證明呢？存疑待考！