如何创建Azure Data Lake Storage到API的连接并实现数据读取

前置准备 #

• 开发环境安装Python ADLS Gen2依赖包，执行命令：pip install azure-storage-file-datalake azure-identity requests
• 提前获取权限配置：Azure AD认证信息（客户端ID、客户端密钥、租户ID），确保对应身份有目标ADLS容器的读取权限

如果你使用的是ADLS Gen1版本，替换依赖为azure-datalake-store，调整对应客户端初始化逻辑即可，数据读取、API推送的核心流程没有变化。

完整可运行示例代码 #

from azure.identity import DefaultAzureCredential
from azure.storage.filedatalake import DataLakeServiceClient
import requests
import json

# 1. 初始化ADLS客户端
account_name = "你的ADLS存储账户名"
file_system_name = "目标容器名"
file_path = "容器内目标文件的路径，比如xxx/xxx/data.csv"

credential = DefaultAzureCredential()
service_client = DataLakeServiceClient(
    account_url=f"https://{account_name}.dfs.core.windows.net",
    credential=credential
)
file_system_client = service_client.get_file_system_client(file_system=file_system_name)
file_client = file_system_client.get_file_client(file_path)

# 2. 从ADLS读取数据
download = file_client.download_file()
file_content = download.readall()
# 这里可以根据文件格式做解析，以下为CSV转JSON示例，可按实际业务替换解析逻辑
# import pandas as pd
# from io import BytesIO
# df = pd.read_csv(BytesIO(file_content))
# processed_data = df.to_dict(orient="records")
processed_data = {"data": file_content.decode("utf-8")} # 示例数据结构，按实际需要修改

# 3. 推送数据到目标API
api_url = "你的目标API接口地址"
headers = {"Content-Type": "application/json"}
response = requests.post(api_url, data=json.dumps(processed_data), headers=headers, timeout=30)

# 4. 结果校验
if response.status_code in [200, 201]:
    print("数据推送成功")
else:
    print(f"数据推送失败，状态码：{response.status_code}，错误信息：{response.text}")

逻辑说明 #

• 认证环节使用DefaultAzureCredential会自动适配多种场景：本地开发时会读取Azure CLI/VS Code Azure插件的登录信息，生产环境部署到Azure服务时会自动读取托管身份的凭证，不需要硬编码密钥，安全性更高
• 大文件拉取可以使用分块下载逻辑，避免内存溢出：可以通过download_file(offset=xxx, length=xxx)方法分批次读取文件内容，分批次推送到API
• 如果需要拉取整个目录下的所有文件，可以调用file_system_client.get_paths()遍历目录下所有文件，循环执行读取+推送逻辑

异常处理建议 #

• 增加重试机制：对ADLS读取、API调用的超时、5xx类错误添加重试逻辑，可使用tenacity库快速实现
• 增加日志记录：对读取的文件大小、API调用的入参出参、错误信息做日志留存，方便问题排查

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