webhdfs 详解

简介

hdfs 提供了一种除了通过 rpc 的方式进行文件操作的方式之外，还提供了 http 的方式对文件进行操作的方式：webhdfs。支持 HDFS 的完整 FileSystem / FileContext 接口。

其中 Router 和 NameNode 都支持了 webhdfs 的功能，具体实现有差别。

使用

文件系统 URI 与 HTTP URL

WebHDFS 的文件系统方案为“ webhdfs：// ”。WebHDFS 文件系统 URI 具有以下格式。

webhdfs://<主机>:<HTTP_PORT>/<PATH>

上面的 WebHDFS URI 对应于下面的 HDFS URI。

 hdfs://<主机>:<RPC_PORT>/<PATH>

在 REST API 中，在路径中插入前缀“ /webhdfs/v1 ”，并在末尾附加查询。因此，相应的 HTTP URL 具有以下格式。

http://<主机>:<HTTP_PORT>/webhdfs/v1/<PATH>?op=create

详细可以参考：https://hadoop.apache.org/docs/r3.4.1/hadoop-project-dist/hadoop-hdfs/WebHDFS.html

源码实现分析

NameNode webhdfs 源码实现分析

启动和初始化

在 NameNode 启动过程中，启动 NameNode 的 http 模块的时候，启动了 NameNode 的 webhdfs 模块。核心入口函数(NameNodeHttpServer.java)：

void start() throws IOException {
//...
initWebHdfs(conf, bindAddress.getHostName(), httpKeytab, httpServer,
        NamenodeWebHdfsMethods.class.getPackage().getName());
//...

}

从上面代码可以看出 webhdfs 的核心处理类为 NamenodeWebHdfsMethods.java。当前类是每个请求都是由一个 NamenodeWebHdfsMethods 对象处理的，在处理每个请求的时候，需要做下面的初始化：

 public NamenodeWebHdfsMethods(@Context HttpServletRequest request) {
    // the request object is a proxy to thread-locals so we have to extract
    // what we want from it since the external call will be processed in a
    // different thread.
    scheme = request.getScheme();
    userPrincipal = request.getUserPrincipal();
    // get the remote address, if coming in via a trusted proxy server then
    // the address with be that of the proxied client
    remoteAddr = JspHelper.getRemoteAddr(request);
    remotePort = JspHelper.getRemotePort(request);
    supportEZ =
        Boolean.valueOf(request.getHeader(WebHdfsFileSystem.EZ_HEADER));
  }

主要获取了当前登录的用户的相关信息，hdfs 的 nameService 以及是否开启 EC。

请求处理

NamenodeWebHdfsMethods 里面定义的请求类型主要是：

• PUT 请求：主要处理写入类型的求情。

  • CREATE 操作。
  • MKDIRS 操作。
  • CREATESYMLINK 操作。
  • RENAME 操作。
  • SETREPLICATION 操作。
  • SETOWNER 操作。
  • .....

• DELETE 请求：主要处理删除类的请求。主要包含：

  • DELETE 操作：
  • DELETESNAPSHOT 操作：

• GET 请求：主要处理查询类的请求。

• POST 请求：主要处理写入类的请求。主要包含：

  • APPEND 操作。
  • CONCAT 操作。
  • TRUNCATE 操作。
  • UNSETSTORAGEPOLICY 操作。
  • UNSETECPOLICY 操作。

定义参考如下：

 @GET
  @Path("{" + UriFsPathParam.NAME + ":.*}")
  @Produces({MediaType.APPLICATION_OCTET_STREAM + "; " + JettyUtils.UTF_8,
      MediaType.APPLICATION_JSON + "; " + JettyUtils.UTF_8})
  public Response get(
      @Context final UserGroupInformation ugi,
      @QueryParam(DelegationParam.NAME) @DefaultValue(DelegationParam.DEFAULT)
          final DelegationParam delegation,
     //...
      ) throws IOException, InterruptedException {

    init(ugi, delegation, username, doAsUser, path, op, offset, length,
        renewer, bufferSize, xattrEncoding, excludeDatanodes, fsAction,
        snapshotName, oldSnapshotName, tokenKind, tokenService, startAfter);

    return doAs(ugi, new PrivilegedExceptionAction<Response>() {
      @Override
      public Response run() throws IOException, URISyntaxException {
        return get(ugi, delegation, username, doAsUser, path.getAbsolutePath(),
            op, offset, length, renewer, bufferSize, xattrNames, xattrEncoding,
            excludeDatanodes, fsAction, snapshotName, oldSnapshotName,
            tokenKind, tokenService, noredirect, startAfter);
      }
    });
  }

webhdfs 操作实现

CREATE 操作

未完待续