如何使用Azure SDKv2将DataFrame保存为CSV到指定Blob存储容器

需求：将Pandas DataFrame保存到指定私有Blob容器（URL：https://mycontainer1234.blob.core.windows.net/files），该容器无公共访问权限。此前在SDK v1中通过Target Datastore实现，SDK v2已移除该功能，同时希望了解如何通过uri_file/uri_folder创建尚未存在的文件。

首先创建示例DataFrame：

import pandas as pd
import random

# 创建示例数据
data = {
    "Name": [f"Person {i}" for i in range(1, 51)],
    "Age": [random.randint(18, 60) for _ in range(50)]
}

# 创建DataFrame
df = pd.DataFrame(data)

方法1：使用azure-storage-blob SDK直接上传 #

这是最直接的方案，无需依赖Azure ML SDK额外组件，直接操作Blob存储。

步骤1：安装依赖 #

pip install azure-storage-blob pandas

步骤2：实现上传代码 #

支持多种认证方式，以下为常用两种：

方式A：使用Blob容器连接字符串

from azure.storage.blob import BlobServiceClient
import io

# 替换为你的Blob容器连接字符串
connection_string = "你的容器连接字符串"
# 容器名称（从目标URL提取：files）
container_name = "files"
# 容器内目标文件路径
blob_name = "people_data.csv"

# 初始化客户端
blob_service_client = BlobServiceClient.from_connection_string(connection_string)
container_client = blob_service_client.get_container_client(container_name)

# 将DataFrame转为字节流（以CSV为例，也支持Parquet等格式）
csv_data = df.to_csv(index=False).encode('utf-8')
stream = io.BytesIO(csv_data)

# 上传文件（overwrite=True覆盖同名文件）
container_client.upload_blob(name=blob_name, data=stream, overwrite=True)

方式B：使用Azure AD认证（生产环境推荐）

适用于Managed Identity、Service Principal等场景，无需硬编码密钥：

from azure.storage.blob import BlobServiceClient
from azure.identity import DefaultAzureCredential
import io

# 从目标URL提取存储账户名称
account_name = "mycontainer1234"
container_name = "files"
blob_name = "people_data.csv"

# 自动获取凭据（本地开发、云环境均兼容）
credential = DefaultAzureCredential()

# 初始化客户端
blob_service_client = BlobServiceClient(
    account_url=f"https://{account_name}.blob.core.windows.net",
    credential=credential
)
container_client = blob_service_client.get_container_client(container_name)

# 转换并上传DataFrame
csv_data = df.to_csv(index=False).encode('utf-8')
stream = io.BytesIO(csv_data)
container_client.upload_blob(name=blob_name, data=stream, overwrite=True)

方法2：结合Azure ML SDK v2的uri_file/uri_folder实体 #

在Azure ML管道中，uri_file/uri_folder用于声明存储路径，需配合存储SDK完成文件创建。以下是管道组件中的实现示例：

from azure.ai.ml import MLClient, command
from azure.ai.ml.entities import UriFile
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient
from urllib.parse import urlparse
import pandas as pd
import random
import io

# 初始化MLClient
ml_client = MLClient(
    DefaultAzureCredential(),
    subscription_id="你的订阅ID",
    resource_group_name="你的资源组",
    workspace_name="你的工作区"
)

# 定义管道组件
@command(
    outputs={
        "output_blob": UriFile(path="https://mycontainer1234.blob.core.windows.net/files/people_data.csv")
    }
)
def save_df_to_blob(output_blob):
    # 创建示例DataFrame（实际场景可传入已有df）
    data = {
        "Name": [f"Person {i}" for i in range(1, 51)],
        "Age": [random.randint(18, 60) for _ in range(50)]
    }
    df = pd.DataFrame(data)

    # 解析UriFile路径中的存储信息
    parsed_url = urlparse(output_blob)
    account_name = parsed_url.netloc.split('.')[0]
    container_name = parsed_url.path.lstrip('/').split('/')[0]
    blob_name = '/'.join(parsed_url.path.lstrip('/').split('/')[1:])

    # 认证并上传文件
    credential = DefaultAzureCredential()
    blob_service_client = BlobServiceClient(
        account_url=f"https://{account_name}.blob.core.windows.net",
        credential=credential
    )
    container_client = blob_service_client.get_container_client(container_name)

    csv_data = df.to_csv(index=False).encode('utf-8')
    stream = io.BytesIO(csv_data)
    container_client.upload_blob(name=blob_name, data=stream, overwrite=True)

# 提交管道作业
job = save_df_to_blob()
returned_job = ml_client.jobs.create_or_update(job)
ml_client.jobs.stream(returned_job.name)

关键说明 #

• UriFile(path="xxx")可预先定义不存在的文件路径，作为管道输出声明。
• 组件内部通过解析路径提取存储信息，结合Azure存储SDK完成文件创建与上传。
• 私有容器的认证由DefaultAzureCredential自动处理，兼容多种环境。

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