如何使用PySpark/Databricks dbutils获取S3目录下的文件名列表

核心解决方法 #

dbutils.fs.ls(MYPATH)返回的结果是FileInfo对象列表，每个对象包含path（文件完整路径）、name（文件名）、size（文件大小，目录大小为0）三个属性，直接提取path字段即可构造你需要的fileList数组。

基础版本（仅处理当前目录下的文件，忽略子目录）

# 自定义参数配置
MYPATH = "你的S3桶相对路径"
jsonDir = "json文件存储目标路径"
otherDir = "非json文件存储目标路径"

# 构造fileList：提取当前目录下所有文件的完整路径
fileList = [file_info.path for file_info in dbutils.fs.ls(MYPATH) if file_info.size > 0]

# 提前创建目标目录避免移动报错
dbutils.fs.mkdirs(jsonDir)
dbutils.fs.mkdirs(otherDir)

# 文件分流移动逻辑
for file_path in fileList:
    if file_path.endswith(".json"):
        dbutils.fs.mv(file_path, jsonDir)
    else:
        dbutils.fs.mv(file_path, otherDir)

进阶版本（递归处理所有层级子目录下的文件）

如果需要遍历MYPATH下所有子目录的文件统一分流，使用递归方法获取全量文件路径即可：

# 自定义参数配置
MYPATH = "你的S3桶相对路径"
jsonDir = "json文件存储目标路径"
otherDir = "非json文件存储目标路径"

# 递归遍历所有目录获取文件路径
def get_all_file_paths(base_path):
    all_files = []
    for item in dbutils.fs.ls(base_path):
        if item.size > 0:
            all_files.append(item.path)
        else:
            all_files.extend(get_all_file_paths(item.path))
    return all_files

fileList = get_all_file_paths(MYPATH)

# 提前创建目标目录避免移动报错
dbutils.fs.mkdirs(jsonDir)
dbutils.fs.mkdirs(otherDir)

# 文件分流移动逻辑
for file_path in fileList:
    if file_path.endswith(".json"):
        dbutils.fs.mv(file_path, jsonDir)
    else:
        dbutils.fs.mv(file_path, otherDir)

注意：如果你的路径中存在名称带.json的目录，上述逻辑会自动过滤（因为目录size为0不会进入fileList），不会误移动。如果需要移动目录本身，去掉列表生成式中的if file_info.size > 0过滤条件即可。

内容的提问来源于stack exchange，提问作者Timothy Clotworthy