如何用Python遍历ADLS文件夹？适配现有DBFS遍历逻辑

完全可以实现，核心是通过Azure官方存储SDK访问ADLS Gen2文件系统，再结合Pandas完成数据读取与合并。以下是具体实现方案：

1. 安装必要依赖 #

先安装Azure存储服务的Python SDK及相关依赖：

pip install azure-storage-file-datalake pandas regex

2. 配置ADLS连接信息 #

提前准备好ADLS的认证信息，常用两种认证方式：

• 账号密钥认证：需ADLS存储账户名、账户密钥、目标文件系统名（即ADLS中的容器）
• SAS Token认证：需存储账户名、SAS Token、目标文件系统名

3. 递归遍历ADLS目录并处理文件 #

ADLS没有直接对应os.walk的API，需自行实现递归遍历逻辑。通过DataLakeServiceClient建立连接后，逐层遍历目录，匹配文件名后读取CSV数据。

完整代码示例 #

import pandas as pd
import re
from azure.storage.filedatalake import DataLakeServiceClient

# -------------------------- 配置参数 --------------------------
ADLS_ACCOUNT_NAME = "你的ADLS存储账户名"
ADLS_ACCOUNT_KEY = "你的ADLS账户密钥"  # 用SAS的话替换为SAS_TOKEN变量
FILE_SYSTEM_NAME = "目标文件系统名（容器名）"
ROOT_DIR = "data"  # 对应原代码中的/dbfs/data目录
contrast = "你的文件名匹配正则表达式"
contrast_rolled = "你的文件名匹配正则表达式"

# -------------------------- 初始化结果DataFrame --------------------------
tot_contrast_sh = pd.DataFrame()
tot_contrast_rolled_sh = pd.DataFrame()

# -------------------------- 连接ADLS服务 --------------------------
# 方式1：账号密钥连接
service_client = DataLakeServiceClient(
    account_url=f"https://{ADLS_ACCOUNT_NAME}.dfs.core.windows.net",
    credential=ADLS_ACCOUNT_KEY
)

# 方式2：SAS Token连接（替换上面的连接代码即可）
# service_client = DataLakeServiceClient(
#     account_url=f"https://{ADLS_ACCOUNT_NAME}.dfs.core.windows.net",
#     credential="你的SAS Token"
# )

file_system_client = service_client.get_file_system_client(file_system=FILE_SYSTEM_NAME)

# -------------------------- 递归遍历目录 --------------------------
def traverse_directory(dir_client):
    global tot_contrast_sh, tot_contrast_rolled_sh
    
    # 获取当前目录下的所有子目录和文件
    paths = dir_client.list_paths()
    for path in paths:
        if path.is_directory:
            # 递归遍历子目录
            subdir_client = dir_client.get_sub_directory_client(path.name)
            traverse_directory(subdir_client)
        else:
            # 处理文件
            file_name = path.name.split("/")[-1]
            if re.search(contrast, file_name):
                # 读取CSV文件流到DataFrame
                file_client = dir_client.get_file_client(path.name)
                download_stream = file_client.download_file()
                df = pd.read_csv(download_stream)
                tot_contrast_sh = pd.concat([tot_contrast_sh, df], ignore_index=True)
            elif re.search(contrast_rolled, file_name):
                file_client = dir_client.get_file_client(path.name)
                download_stream = file_client.download_file()
                df = pd.read_csv(download_stream)
                tot_contrast_rolled_sh = pd.concat([tot_contrast_rolled_sh, df], ignore_index=True)

# 启动根目录遍历
root_dir_client = file_system_client.get_directory_client(ROOT_DIR)
traverse_directory(root_dir_client)

# 输出结果预览
print("匹配contrast的合并数据：")
print(tot_contrast_sh.head())
print("\n匹配contrast_rolled的合并数据：")
print(tot_contrast_rolled_sh.head())

关键说明 #

• 替代os.walk：通过递归函数traverse_directory遍历ADLS目录结构，list_paths方法返回当前目录下的所有子目录和文件。
• 文件读取：通过download_file()获取文件流，直接传入pd.read_csv，无需下载到本地磁盘，提升效率。
• 数据合并：用pd.concat替代已弃用的append方法，符合Pandas新版本规范。

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