PROSAGA码农传奇-HDFS-Python：如何从HDFS导入目录中的文件列表

<div class =“post-text”itemprop =“text”>
  <P>
    的<strong>
      Declaimer
    </强>
    ：这将是漫长而乏味的。但考虑到这种情况，我会尽量使其尽可能通用且可重复。
  </p>
  <HR />
  <P>
    鉴于没有外部库的要求（除了
     <code>
      pandas
    </code>
    ？），没有必要选择。我建议使用
     <code>
      WebHDFS
    </code>
     越多越好。
  </p>
  <P>
    AFAIK，安装
    <EM>
      HDFS
    </EM>
    默认情况下，包括安装
    <EM>
      WebHDFS
    </EM>
    。以下解决方案严重依赖
    <EM>
      WebHDFS
    </EM>
    。
  </p>
  <H3>
    第一步
  </H3>
  <P>
    首先，你必须要注意
    <EM>
      WebHDFS
    </EM>
     网址。
    <EM>
      WebHDFS
    </EM>
     已安装
    的<strong>
      HDFS Namenode（s）
    </强>
    ，默认端口是
    的<strong>
      50070
    </强>
    。
  </p>
  <P>
    因此，我们从一开始
     <code>
      http://[namenode_ip]:50070/webhdfs/v1/
    </code>
    ，哪里
     <code>
      /webhdfs/v1
    </code>
    /是所有人的共同网址。
  </p>
  <P>
    为了举例，我们假设它为
     <code>
      http://192.168.10.1:50070/web/hdfs/v1
    </code>
    。
  </p>
  <H3>
    第二步
  </H3>
  <P>
    通常，人们可以使用
     <code>
      curl
    </code>
     列出HDFS目录的内容。
有关详细说明，请参阅
    <a href="https://hadoop.apache.org/docs/r1.0.4/webhdfs.html#LISTSTATUS" rel="nofollow noreferrer">
      WebHDFS REST API：列出目录
    </A>
  </p>
  <P>
    如果你要使用
     <code>
      curl
    </code>
    ，以下提供
     <code>
      FileStatuses
    </code>
     给定目录中的所有文件。
  </p>
   <pre>
    <code>
      curl "http://192.168.10.1:50070/webhdfs/v1/<PATH>?op=LISTSTATUS"
             ^^^^^^^^^^^^ ^^^^^             ^^^^  ^^^^^^^^^^^^^
             Namenode IP  Port              Path  Operation

</code>
  </pre>
  <P>
    如上所述，这将返回JSON对象中的FileStatuses：
  </p>
   <pre>
    <code>
      {
  "FileStatuses":
  {
    "FileStatus":
    [
      {
        "accessTime"      : 1320171722771,
        "blockSize"       : 33554432,
        "group"           : "supergroup",
        "length"          : 24930,
        "modificationTime": 1320171722771,
        "owner"           : "webuser",
        "pathSuffix"      : "a.patch",
        "permission"      : "644",
        "replication"     : 1,
        "type"            : "FILE"
      },
      {
        "accessTime"      : 0,
        "blockSize"       : 0,
        "group"           : "supergroup",
        "length"          : 0,
        "modificationTime": 1320895981256,
        "owner"           : "szetszwo",
        "pathSuffix"      : "bar",
        "permission"      : "711",
        "replication"     : 0,
        "type"            : "DIRECTORY"
      },
      ...
    ]
  }
}

</code>
  </pre>
  <P>
    使用python的默认库可以实现相同的结果：
  </p>
   <pre>
    <code>
      import requests

my_path = '/my_path/'
curl = requests.get('http://192.168.10.1:50070/webhdfs/v1/%s?op=LISTSTATUS' % my_path)

</code>
  </pre>
  <P>
    如上所示，每个文件的实际状态比结果JSON低两级。换句话说，要获取每个文件的FileStatus：
  </p>
   <pre>
    <code>
      curl.json()['FileStatuses']['FileStatus']

[
  {
    "accessTime"      : 1320171722771,
    "blockSize"       : 33554432,
    "group"           : "supergroup",
    "length"          : 24930,
    "modificationTime": 1320171722771,
    "owner"           : "webuser",
    "pathSuffix"      : "a.patch",
    "permission"      : "644",
    "replication"     : 1,
    "type"            : "FILE"
  },
  {
    "accessTime"      : 0,
    "blockSize"       : 0,
    "group"           : "supergroup",
    "length"          : 0,
    "modificationTime": 1320895981256,
    "owner"           : "szetszwo",
    "pathSuffix"      : "bar",
    "permission"      : "711",
    "replication"     : 0,
    "type"            : "DIRECTORY"
  },
  ...
]

</code>
  </pre>
  <H3>
    第三步
  </H3>
  <P>
    既然您现在拥有了所需的所有信息，那么您需要做的就是解析。
  </p>
   <pre>
    <code>
      import os

file_paths = []
for file_status in curl.json()['FileStatuses']['FileStatus']:
    file_name = file_status['pathSuffix']
    # this is the file name in the queried directory
    if file_name.endswith('.csv'):
    # if statement is only required if the directory contains unwanted files (i.e. non-csvs).
        file_paths.append(os.path.join(path, file_name))
        # os.path.join asserts your result consists of absolute path

file_paths
['/my_path/file1.csv',
 '/my_path/file2.csv',
 ...]

</code>
  </pre>
  <H3>
    最后一步
  </H3>
  <P>
    现在你知道了文件和WebHDFS链接的路径，
     <code>
      pandas.read_csv
    </code>
     可以处理其余的工作。
  </p>
   <pre>
    <code>
      import pandas as pd

dfs = []
web_url = "http://192.168.10.1:50070/webhdfs/v1/%s?op=OPEN"
#                                                  ^^^^^^^
#                                    Operation is now OPEN
for file_path in file_paths:
    file_url = web_url % file_path
    # http://192.168.10.1:50070/webhdfs/v1/my_path/file1.csv?op=OPEN
    dfs.append(pd.read_csv(file_url))

</code>
  </pre>
  <P>
    在那里你和所有人一起去
     <code>
      .csv
    </code>
    导入并分配给
     <code>
      dfs
    </code>
    。
  </p>
  <H3>
    警告
  </H3>
  <P>
    如果您的HDFS配置为
    的<strong>
      哈
    </强>
     （高可用性），会有多个
    的<strong>
      namenodes
    </强>
     因此你的
     <code>
      namenode_ip
    </code>
     必须相应地设置：它必须是活动节点的IP。
  </p>
</DIV>