Azure 存储文件数据湖 Python (SDK)
Azure Data Lake Storage Gen2 Python SDK
适用于大数据分析工作负载的分层文件系统。
安装
pip install azure-storage-file-datalake azure-identity环境变量
AZURE_STORAGE_ACCOUNT_URL=https://<account>.dfs.core.windows.net身份验证
from azure.identity import DefaultAzureCredential
from azure.storage.filedatalake import DataLakeServiceClient
credential = DefaultAzureCredential()
account_url = "https://<account>.dfs.core.windows.net"
service_client = DataLakeServiceClient(account_url=account_url, credential=credential)
客户端层级
| 客户端 | 用途 |
|--------|---------|
| DataLakeServiceClient | 账户级操作 |
| FileSystemClient | 容器（文件系统）操作 |
| DataLakeDirectoryClient | 目录操作 |
| DataLakeFileClient | 文件操作 |
文件系统操作
# 创建文件系统 (容器)
file_system_client = service_client.create_file_system("myfilesystem")
获取现有文件系统
file_system_client = service_client.get_file_system_client("myfilesystem")
删除
service_client.delete_file_system("myfilesystem")
列出文件系统
for fs in service_client.list_file_systems():
print(fs.name)目录操作
file_system_client = service_client.get_file_system_client("myfilesystem")
创建目录
directory_client = file_system_client.create_directory("mydir")
创建嵌套目录
directory_client = file_system_client.create_directory("path/to/nested/dir")
获取目录客户端
directory_client = file_system_client.get_directory_client("mydir")
删除目录
directory_client.delete_directory()
重命名/移动目录
directory_client.rename_directory(new_name="myfilesystem/newname")文件操作
上传文件
# 获取文件客户端
file_client = file_system_client.get_file_client("path/to/file.txt")
从本地文件上传
with open("local-file.txt", "rb") as data:
file_client.upload_data(data, overwrite=True)
上传字节流
file_client.upload_data(b"Hello, Data Lake!", overwrite=True)
追加数据 (适用于大文件)
file_client.append_data(data=b"chunk1", offset=0, length=6)
file_client.append_data(data=b"chunk2", offset=6, length=6)
file_client.flush_data(12) # 提交数据下载文件
file_client = file_system_client.get_file_client("path/to/file.txt")
下载全部内容
download = file_client.download_file()
content = download.readall()
下载到文件
with open("downloaded.txt", "wb") as f:
download = file_client.download_file()
download.readinto(f)
下载指定范围
download = file_client.download_file(offset=0, length=100)删除文件
file_client.delete_file()列出内容
# 列出路径 (文件和目录)
for path in file_system_client.get_paths():
print(f"{'DIR' if path.is_directory else 'FILE'}: {path.name}")
列出目录中的路径
for path in file_system_client.get_paths(path="mydir"):
print(path.name)
递归列出
for path in file_system_client.get_paths(path="mydir", recursive=True):
print(path.name)文件/目录属性
获取属性
properties = file_client.get_file_properties() print(f"大小: {properties.size}") print(f"最后修改时间: {properties.last_modified}")设置元数据
file_client.set_metadata(metadata={"processed": "true"})## 访问控制 (ACL)获取 ACL
acl = directory_client.get_access_control() print(f"所有者: {acl['owner']}") print(f"权限: {acl['permissions']}")设置 ACL
directory_client.set_access_control( owner="user-id", permissions="rwxr-x---" )更新 ACL 条目
from azure.storage.filedatalake import AccessControlChangeResult directory_client.update_access_control_recursive( acl="user:user-id:rwx" )## 异步客户端async def datalake_operations():
credential = DefaultAzureCredential()
async with DataLakeServiceClient(
account_url="https://<account>.dfs.core.windows.net",
credential=credential
) as service_client:
file_system_client = service_client.get_file_system_client("myfilesystem")
file_client = file_system_client.get_file_client("test.txt")
await file_client.upload_data(b"async content", overwrite=True)
download = await file_client.download_file()
content = await download.readall()
import asyncio
asyncio.run(datalake_operations())
``
最佳实践
1. 使用分层命名空间 以实现文件系统语义
2. 使用 append_data + flush_data 进行大文件上传
3. 在目录级别设置 ACL 并由子项继承
4. 在高吞吐量场景下使用异步客户端
5. 使用 get_paths 并设置 recursive=True` 以列出完整目录
6. 设置元数据 以定义自定义文件属性
7. 对于简单的对象存储用例，可考虑使用 Blob API
适用场景
此技能适用于执行概览中所描述的工作流或操作。局限性
- 仅在任务与上述描述的范围明确匹配时使用此技能。
- 不要将输出视为针对特定环境的验证、测试或专家评审的替代方案。
- 如果缺少必要的输入、权限、安全边界或成功标准，请停止并请求澄清。
