Pyspark：获取 HDFS 路径上的文件/目录列表答案

【问题标题】：Pyspark: get list of files/directories on HDFS pathPyspark：获取 HDFS 路径上的文件/目录列表
【发布时间】：2016-06-15 12:08:59
【问题描述】：

根据标题。我知道textFile，但顾名思义，它只适用于文本文件。我需要访问 HDFS 或本地路径上的路径内的文件/目录。我正在使用 pyspark。

【问题讨论】：

标签： hadoop apache-spark pyspark

【解决方案1】：

使用 JVM 网关可能不是那么优雅，但在某些情况下，下面的代码可能会有所帮助：

URI           = sc._gateway.jvm.java.net.URI
Path          = sc._gateway.jvm.org.apache.hadoop.fs.Path
FileSystem    = sc._gateway.jvm.org.apache.hadoop.fs.FileSystem
Configuration = sc._gateway.jvm.org.apache.hadoop.conf.Configuration


fs = FileSystem.get(URI("hdfs://somehost:8020"), Configuration())

status = fs.listStatus(Path('/some_dir/yet_another_one_dir/'))

for fileStatus in status:
    print(fileStatus.getPath())

【讨论】：

如果要过滤结果，请使用 globStatus 而不是 fileStatus，例如status = fs.globStatus(Path('/some_dir/yet_another_one_dir/*.csv'))
这很好，因为它不需要我上传额外的库来 spark-submit。
在pyspark中获取/查找somehost，即namenode的好方法是什么？
我有数千个文件，这段代码files = [file.getPath() for file in status] 需要一段时间。正常吗？我会说这不是最有效的方法。
@jcomeau_ictx 我想你的意思是说使用globStatus 而不是listStatus，而不是fileStatus（这只是一个临时变量）。

【解决方案2】：

我认为将 Spark 仅视为一种数据处理工具会很有帮助，它的域从加载数据开始。它可以读取多种格式，并且支持 Hadoop glob 表达式，这对于从 HDFS 中的多个路径读取非常有用，但它没有我知道的用于遍历目录或文件的内置工具，也没有专用于与 Hadoop 或 HDFS 交互的实用程序。

有一些可用的工具可以满足您的需求，包括esutil 和hdfs。 hdfs lib 支持 CLI 和 API，您可以直接跳转到“我如何在 Python 中列出 HDFS 文件”here。它看起来像这样：

from hdfs import Config
client = Config().get_client('dev')
files = client.list('the_dir_path')

【讨论】：

嗨，你能指导我如何制作 hdfscli.cfg 文件，我不知道该放什么端口号。 [全局] default.alias = dev [dev.alias] url = dev.namenode:port user = ann
第二个@ShivamKotwalia，因为我不能指定一个用户访问我的 EMR，它必须是完全动态的，并且我没有通过我的代码传递配置的用户
我支持放入get_client('dev')，我没有别名'dev'
@HahaTTpro 您可以在没有参数的情况下调用该函数，仅当您设置了备用别名时才需要。如果不确定，请查找文件 ~/.hdfscli.cfg，然后在其中搜索 default.alias

【解决方案3】：

如果你使用PySpark，你可以交互式地执行命令：

列出所选目录中的所有文件：

hdfs dfs -ls <path> 例如：hdfs dfs -ls /user/path：

import os
import subprocess

cmd = 'hdfs dfs -ls /user/path'
files = subprocess.check_output(cmd, shell=True).strip().split('\n')
for path in files:
  print path

或在所选目录中搜索文件：

hdfs dfs -find <path> -name <expression> 例如：hdfs dfs -find /user/path -name *.txt：

import os
import subprocess

cmd = 'hdfs dfs -find {} -name *.txt'.format(source_dir)
files = subprocess.check_output(cmd, shell=True).strip().split('\n')
for path in files:
  filename = path.split(os.path.sep)[-1].split('.txt')[0]
  print path, filename

【讨论】：

嗨，文件 = subprocess.check_output(cmd_find).strip().split('\n') 不应该是 files = subprocess.check_output(cmd).strip().split(' \n') 我尝试编辑，但 SO 说编辑必须大于 6 个更改。
@Darius Morawiec: 如何执行hdfs dfs -rm -r 命令？是使用相同的 check_output 方法还是其他方式？
@Shankar，为此您可以使用subprocess.call 或subprocess.check_call。

【解决方案4】：

如果你想读入一个目录下的所有个文件，查看sc.wholeTextFiles[doc]，但是注意文件的内容读入的是单行的值，大概是不是想要的结果。

如果您只想读取一些文件，则生成路径列表（使用普通 hdfs ls 命令加上您需要的任何过滤）并将其传递到 sqlContext.read.text [doc] 然后从 DataFrame 转换为RDD 似乎是最好的方法。

【讨论】：

【解决方案5】：

这可能对你有用：

import subprocess, re
def listdir(path):
    files = str(subprocess.check_output('hdfs dfs -ls ' + path, shell=True))
    return [re.search(' (/.+)', i).group(1) for i in str(files).split("\\n") if re.search(' (/.+)', i)]

listdir('/user/')

这也有效：

hadoop = sc._jvm.org.apache.hadoop
fs = hadoop.fs.FileSystem
conf = hadoop.conf.Configuration()
path = hadoop.fs.Path('/user/')
[str(f.getPath()) for f in fs.get(conf).listStatus(path)]

【讨论】：

【解决方案6】：

使用snakebite 库有一个简单的方法来做到这一点

from snakebite.client import Client

hadoop_client = Client(HADOOP_HOST, HADOOP_PORT, use_trash=False)

for x in hadoop_client.ls(['/']):

...     print x

【讨论】：