我尝试从python中的HDFS导入文件列表。
如何从HDFS执行此操作:
path = r’/ my_path’allFiles = glob.glob(path +“/ *。csv”)
df_list = []for file_ in allFiles: df = pd.read_csv(文件_,…
的 Declaimer 强> :这将是漫长而乏味的。但考虑到这种情况,我会尽量使其尽可能通用且可重复。
鉴于没有外部库的要求(除了 pandas ?),没有必要选择。我建议使用 WebHDFS 越多越好。
pandas
WebHDFS
AFAIK,安装 HDFS 默认情况下,包括安装 WebHDFS 。以下解决方案严重依赖 WebHDFS 。
首先,你必须要注意 WebHDFS 网址。 WebHDFS 已安装 的 HDFS Namenode(s) 强> ,默认端口是 的 50070 强> 。
因此,我们从一开始 http://[namenode_ip]:50070/webhdfs/v1/ ,哪里 /webhdfs/v1 /是所有人的共同网址。
http://[namenode_ip]:50070/webhdfs/v1/
/webhdfs/v1
为了举例,我们假设它为 http://192.168.10.1:50070/web/hdfs/v1 。
http://192.168.10.1:50070/web/hdfs/v1
通常,人们可以使用 curl 列出HDFS目录的内容。 有关详细说明,请参阅 WebHDFS REST API:列出目录
curl
如果你要使用 curl ,以下提供 FileStatuses 给定目录中的所有文件。
FileStatuses
curl "http://192.168.10.1:50070/webhdfs/v1/<PATH>?op=LISTSTATUS" ^^^^^^^^^^^^ ^^^^^ ^^^^ ^^^^^^^^^^^^^ Namenode IP Port Path Operation
如上所述,这将返回JSON对象中的FileStatuses:
{ "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" }, ... ] } }
使用python的默认库可以实现相同的结果:
import requests my_path = '/my_path/' curl = requests.get('http://192.168.10.1:50070/webhdfs/v1/%s?op=LISTSTATUS' % my_path)
如上所示,每个文件的实际状态比结果JSON低两级。换句话说,要获取每个文件的FileStatus:
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" }, ... ]
既然您现在拥有了所需的所有信息,那么您需要做的就是解析。
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', ...]
现在你知道了文件和WebHDFS链接的路径, pandas.read_csv 可以处理其余的工作。
pandas.read_csv
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))
在那里你和所有人一起去 .csv 导入并分配给 dfs 。
.csv
dfs
如果您的HDFS配置为 的 哈 强> (高可用性),会有多个 的 namenodes 强> 因此你的 namenode_ip 必须相应地设置:它必须是活动节点的IP。
namenode_ip