你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站,请访问 https://docs.azure.cn

在交互式开发期间从 Azure 云存储访问数据

适用于:Python SDK azure-ai-ml v2(当前)

机器学习项目通常从探索数据分析(EDA)和数据预处理(清理、特征工程)开始。 它还包括生成机器学习模型原型来验证假设。 此 原型制作 项目阶段具有高度交互性,它适合在 Jupyter 笔记本或 IDE 中使用 Python 交互式控制台进行开发。 在这篇文章中,你将学会如何:

  • 像访问文件系统一样访问 Azure 机器学习数据存储 URI 中的数据。
  • 使用 mltable Python 库将 Azure 机器学习数据资产具体化到 Pandas 中。
  • 使用 azcopy 工具通过显式下载将数据实体化。

Prerequisites

提示

本文中的指南介绍了交互式开发期间的数据访问。 它适用于可以运行 Python 会话的任何主机。 此主机可以是本地计算机、云 VM、GitHub Codespace 或其他主机。 使用 Azure 机器学习计算实例 - 完全托管和预配置的云工作站。 有关详细信息,请参阅创建 Azure 机器学习计算实例

重要说明

确保已在 Python 环境中安装最新的 azure-fsspecmltablePython 库和 azure-ai-ml Python 库:

pip install -U azureml-fsspec==1.3.1 mltable azure-ai-ml

最新的 azure-fsspec 包版本可能会随时间而变化。 有关azure-fsspec包的详细信息,请参阅azureml-fsspec 1.3.1

像访问文件系统一样访问数据存储 URI 中的数据

Azure 机器学习数据存储是对现有 Azure 存储帐户的引用。 创建和使用数据存储的好处包括:

  • 与不同存储类型(Blob、文件、Azure Data Lake Storage)交互的常见易于使用的 API。
  • 在团队运营中更轻松地发现有用的数据存储。
  • 支持基于凭据(例如 SAS 令牌)和基于标识(Microsoft Entra ID 或托管标识)的数据访问。
  • 对于基于凭据的访问,保护连接信息,以避免脚本中的密钥泄露。
  • 可以在 Azure 机器学习工作室 UI 中浏览数据和复制和粘贴数据存储 URI。

数据存储 URI 是统一资源标识符,它是对 Azure 存储帐户中的存储位置(路径)的引用。 数据存储 URI 采用以下格式:

# Azure Machine Learning workspace details:
subscription = '<subscription_ID>'
resource_group = '<resource_group>'
workspace = '<workspace>'
datastore_name = '<datastore>'
path_on_datastore = '<path>'

# Long-form datastore URI format:
uri = f'azureml://subscriptions/{subscription}/resourcegroups/{resource_group}/workspaces/{workspace}/datastores/{datastore_name}/paths/{path_on_datastore}'.

这些数据存储 URI 是文件系统规范 (fsspec) 的已知实现:一种统一的 Python 式接口,适用于本地、远程和嵌入式文件系统及字节存储。 首先,请使用 pip 来安装 azureml-fsspec 包及其依赖项 azureml-dataprep 包。 然后,可以使用 Azure 机器学习数据存储 fsspec 实现。

Azure 机器学习数据存储 fsspec 实现会自动处理 Azure 机器学习数据存储使用的凭据/标识传递。 可以避免脚本中的帐户密钥泄露和计算实例上的额外登录过程。

例如,可以直接在 Pandas 中使用数据存储 URI。 此示例演示了如何读取 CSV 文件:

import pandas as pd

df = pd.read_csv("azureml://subscriptions/<subscription_ID>/resourcegroups/<resource_group_name>/workspaces/<workspace_name>/datastores/<datastore_name>/paths/<folder>/<filename>.csv")
df.head()

提示

若要避免需要记住数据存储 URI 格式,可以按照以下步骤在工作室 UI 中复制和粘贴数据存储 URI:

  1. 在左侧菜单中选择 “数据” ,然后选择“ 数据存储 ”选项卡。
  2. 选择数据存储名称,然后选择“ 浏览”。
  3. 找到要读入 Pandas 的文件或文件夹,选择其旁边的省略号(...),然后选择 “复制 URI”。 然后,可以选择要复制到笔记本或脚本中的数据存储 URI。 显示如何复制数据存储 URI 的屏幕截图。

还可以实例化 Azure 机器学习文件系统来处理类似文件系统的命令,例如lsglobexistsopen

  • ls() 方法会列出特定目录中的文件。 可以使用 ls()ls(.)ls (<folder_level_1>/<folder_level_2>) 列出文件。 在相对路径中,...同时受支持。
  • glob() 方法支持 *** 通配符匹配。
  • 该方法 exists() 返回一个布尔值,该值指示当前根目录中是否存在指定的文件。
  • 该方法 open() 返回一个类似于文件的对象,该对象可以传递给任何其他希望使用 Python 文件的库。 代码还可以使用此对象,就像它是普通的 Python 文件对象一样。 这些类似于文件的对象遵循 with 上下文的用法,如以下示例中所示:
from azureml.fsspec import AzureMachineLearningFileSystem

# Instantiate the filesystem by using the following URI.
fs = AzureMachineLearningFileSystem('azureml://subscriptions/<subscriptionID>/resourcegroups/<resource_group_name>/workspaces/<workspace_name>/datastores/<datastore_name>')

fs.ls() # List folders and files in datastore datastorename.

# Output example:
# folder1
# folder2
# file3.csv

# Use an open context.
with fs.open('./folder1/file1.csv') as f:
    # Do some process.
    process_file(f)

使用 AzureMachineLearningFileSystem 上传文件

from azureml.fsspec import AzureMachineLearningFileSystem
# Instantiate the filesystem by using the following URI.
fs = AzureMachineLearningFileSystem('azureml://subscriptions/<subscriptionID>/resourcegroups/<resource_group_name>/workspaces/<workspace_name>/datastores/<datastore_name>/paths/')

# You can set recursive to False to upload a file.
fs.upload(lpath='data/upload_files/crime-spring.csv', rpath='data/fsspec', recursive=False, **{'overwrite': 'MERGE_WITH_OVERWRITE'})

# You need to set recursive to True to upload a folder.
fs.upload(lpath='data/upload_folder/', rpath='data/fsspec_folder', recursive=True, **{'overwrite': 'MERGE_WITH_OVERWRITE'})

lpath 是本地路径,rpath 是远程路径。 如果指定的 rpath 文件夹不存在,该方法将为你创建文件夹。

该方法支持三 overwrite 种模式:

  • APPEND。 如果目标路径中存在同名的文件,APPEND 将保留原始文件。
  • FAIL_ON_FILE_CONFLICT。 如果目标路径中存在同名的文件,FAIL_ON_FILE_CONFLICT将引发错误。
  • MERGE_WITH_OVERWRITE。 如果目标路径中存在同名的文件,MERGE_WITH_OVERWRITE用新文件覆盖该现有文件。

使用 AzureMachineLearningFileSystem 下载文件

# You can set recursive to False to download a file.
# The downloading overwrite option is determined by the local system. It's MERGE_WITH_OVERWRITE.
fs.download(rpath='data/fsspec/crime-spring.csv', lpath='data/download_files/', recursive=False)

# You need to set recursive to True to download a folder.
fs.download(rpath='data/fsspec_folder', lpath='data/download_folder/', recursive=True)

示例

这些示例演示如何在常见方案中使用文件系统规范。

将单个 CSV 文件读取到 Pandas 中

可以使用以下代码将单个 CSV 文件读入 Pandas:

import pandas as pd

df = pd.read_csv("azureml://subscriptions/<subscription_ID>/resourcegroups/<resource_group_name>/workspaces/<workspace_name>/datastores/<datastore_name>/paths/<folder>/<file_name>.csv")

将 CSV 文件的文件夹读入 Pandas

Pandas read_csv() 方法不支持读取 CSV 文件的文件夹。 若要处理此限制,请使用 glob 获取 CSV 路径,并使用 Pandas concat() 方法将它们连接到数据帧。 下一个代码示例演示如何使用 Azure 机器学习文件系统实现此串联:

import pandas as pd
from azureml.fsspec import AzureMachineLearningFileSystem

# Define the URI. Update <> placeholders.
uri = 'azureml://subscriptions/<subscription_ID>/resourcegroups/<resource_group_name>/workspaces/<workspace_name>/datastores/<datastore_name>'

# Create the filesystem.
fs = AzureMachineLearningFileSystem(uri)

# Append .csv files in the folder to a list.
dflist = []
for path in fs.glob('/<folder>/*.csv'):
    with fs.open(path) as f:
        dflist.append(pd.read_csv(f))

# Concatenate data frames.
df = pd.concat(dflist)
df.head()

将 CSV 文件读入 Dask

此示例展示了如何将 CSV 文件读取到 Dask 数据帧中:

import dask.dd as dd

df = dd.read_csv("azureml://subscriptions/<subscription_ID>/resourcegroups/<resource_group_name>/workspaces/<workspace_name>/datastores/<datastore_name>/paths/<folder>/<file_name>.csv")
df.head()

将包含 Parquet 文件的文件夹读取到 Pandas 中

在 ETL 过程中,Parquet 文件通常写入文件夹。 然后,进程可以发出与 ETL 相关的文件,例如进度、提交和其他元数据。 此示例显示从 ETL 进程创建的文件(以 _ 开头的文件),然后生成 Parquet 数据文件。

显示 Parquet ETL 进程的屏幕截图。

在这些场景中,你仅读取文件夹中的 Parquet 文件。 忽略 ETL 进程文件。 此代码示例展示了如何使用 glob 模式仅读取文件夹中的 Parquet 文件:

import pandas as pd
from azureml.fsspec import AzureMachineLearningFileSystem

# Define the URI. Update <> placeholders.
uri = 'azureml://subscriptions/<subscription_ID>/resourcegroups/<resource_group_name>/workspaces/<workspace_name>/datastores/<datastore_name>'

# Create the filesystem.
fs = AzureMachineLearningFileSystem(uri)

# Append Parquet files in folder to a list.
dflist = []
for path in fs.glob('/<folder>/*.parquet'):
    with fs.open(path) as f:
        dflist.append(pd.read_parquet(f))

# Concatenate data frames.
df = pd.concat(dflist)
df.head()

从 Azure Databricks 文件系统访问数据

文件系统规格 (fsspec) 具有一系列 已知的实现,包括 Databricks Filesystem (dbfs)。

若要访问 dbfs 资源中的数据,则需要:

使用这些值,在计算实例上为 PAT 令牌创建环境变量:

export ADB_PAT=<personal_access_token>

然后,可以访问 Pandas 中的数据,如以下示例所示:

import os
import pandas as pd

pat = os.getenv(ADB_PAT)
path_on_dbfs = '<absolute_path_on_dbfs>' # e.g. /folder/subfolder/file.csv

storage_options = {
    'instance':'adb-<number>.<number>.azuredatabricks.net', 
    'token': pat
}

df = pd.read_csv(f'dbfs://{path_on_dbfs}', storage_options=storage_options)

使用 pillow 读取图像

from PIL import Image
from azureml.fsspec import AzureMachineLearningFileSystem

# Define the URI. Update <> placeholders.
uri = 'azureml://subscriptions/<subscription_ID>/resourcegroups/<resource_group_name>/workspaces/<workspace_name>/datastores/<datastore_name>'

# Create the filesystem.
fs = AzureMachineLearningFileSystem(uri)

with fs.open('/<folder>/<image.jpeg>') as f:
    img = Image.open(f)
    img.show()

PyTorch 自定义数据集示例

在此示例中,你将创建一个 PyTorch 自定义数据集来处理图像。 在此方案中,存在批注文件(CSV 格式),其整体结构如下:

image_path, label
0/image0.png, label0
0/image1.png, label0
1/image2.png, label1
1/image3.png, label1
2/image4.png, label2
2/image5.png, label2

子文件夹将根据所带标签存储这些图像:

/
└── 📁images
    ├── 📁0
    │   ├── 📷image0.png
    │   └── 📷image1.png
    ├── 📁1
    │   ├── 📷image2.png
    │   └── 📷image3.png
    └── 📁2
        ├── 📷image4.png
        └── 📷image5.png

自定义 PyTorch 数据集类必须实现三个函数:__init____len____getitem__,如下所示:

import os
import pandas as pd
from PIL import Image
from torch.utils.data import Dataset

class CustomImageDataset(Dataset):
    def __init__(self, filesystem, annotations_file, img_dir, transform=None, target_transform=None):
        self.fs = filesystem
        f = filesystem.open(annotations_file)
        self.img_labels = pd.read_csv(f)
        f.close()
        self.img_dir = img_dir
        self.transform = transform
        self.target_transform = target_transform

    def __len__(self):
        return len(self.img_labels)

    def __getitem__(self, idx):
        img_path = os.path.join(self.img_dir, self.img_labels.iloc[idx, 0])
        f = self.fs.open(img_path)
        image = Image.open(f)
        f.close()
        label = self.img_labels.iloc[idx, 1]
        if self.transform:
            image = self.transform(image)
        if self.target_transform:
            label = self.target_transform(label)
        return image, label

然后,你可以将数据集进行实例化,如下所示:

from azureml.fsspec import AzureMachineLearningFileSystem
from torch.utils.data import DataLoader

# Define the URI. Update <> placeholders.
uri = 'azureml://subscriptions/<subscription_ID>/resourcegroups/<resource_group_name>/workspaces/<workspace_name>/datastores/<datastore_name>'

# Create the filesystem.
fs = AzureMachineLearningFileSystem(uri)

# Create the dataset.
training_data = CustomImageDataset(
    filesystem=fs,
    annotations_file='/annotations.csv', 
    img_dir='/<path_to_images>/'
)

# Prepare your data for training with DataLoaders.
train_dataloader = DataLoader(training_data, batch_size=64, shuffle=True)

使用 mltable 库将数据实例化为 Pandas 格式

mltable 库还可以帮助你访问云存储中的数据。 若要使用 mltable 将数据读入 Pandas,请使用以下常规格式:

import mltable

# Define a path, folder, or pattern.
path = {
    'file': '<supported_path>'
    # alternatives
    # 'folder': '<supported_path>'
    # 'pattern': '<supported_path>'
}

# Create an mltable from paths.
tbl = mltable.from_delimited_files(paths=[path])
# alternatives
# tbl = mltable.from_parquet_files(paths=[path])
# tbl = mltable.from_json_lines_files(paths=[path])
# tbl = mltable.from_delta_lake(paths=[path])

# Materialize to Pandas.
df = tbl.to_pandas_dataframe()
df.head()

支持的路径

mltable 库支持从不同的路径类型读取表格数据:

位置 示例
本地计算机上的路径 ./home/username/data/my_data
公共 HTTP(S) 服务器上的路径 https://raw.githubusercontent.com/pandas-dev/pandas/main/doc/data/titanic.csv
Azure 存储上的路径 wasbs://<container_name>@<account_name>.blob.core.windows.net/<path>
abfss://<file_system>@<account_name>.dfs.core.windows.net/<path>
一个长格式 Azure 机器学习数据存储 azureml://subscriptions/<subscription_ID>/resourcegroups/<resource_group_name>/workspaces/<workspace_name>/datastores/<name>/paths/<path>

注意

对于 Azure 存储和 Azure 机器学习数据存储上的路径,mltable 使用凭据直通。 如果无权访问基础存储上的数据,则无法访问数据。

文件、文件夹和 glob

mltable 支持从以下对象读取:

  • 文件。 例如,abfss://<file_system>@<account_name>.dfs.core.windows.net/my-csv.csv
  • 文件夹。 例如,abfss://<file_system>@<account_name>.dfs.core.windows.net/my-folder/
  • glob 模式。 例如,abfss://<file_system>@<account_name>.dfs.core.windows.net/my-folder/*.csv
  • 文件、文件夹和/或 glob 模式的组合。

mltable 通过本地和云存储资源的组合以及文件、文件夹和 glob 的组合,将数据具体化为单个数据帧。 例如:

path1 = {
    'file': 'abfss://filesystem@account1.dfs.core.windows.net/my-csv.csv'
}

path2 = {
    'folder': './home/username/data/my_data'
}

path3 = {
    'pattern': 'abfss://filesystem@account2.dfs.core.windows.net/folder/*.csv'
}

tbl = mltable.from_delimited_files(paths=[path1, path2, path3])

支持的文件格式

mltable 支持以下文件格式。

  • 带分隔符的文本(例如 CSV 文件): mltable.from_delimited_files(paths=[path])
  • Parquet:mltable.from_parquet_files(paths=[path])
  • Delta:mltable.from_delta_lake(paths=[path])
  • JSON 行格式: mltable.from_json_lines_files(paths=[path])

示例

读取 CSV 文件

将此代码片段中的占位符 (<>) 替换为值:

import mltable

path = {
    'file': 'abfss://<filesystem>@<account>.dfs.core.windows.net/<folder>/<file_name>.csv'
}

tbl = mltable.from_delimited_files(paths=[path])
df = tbl.to_pandas_dataframe()
df.head()

读取文件夹中的 Parquet 文件

此示例演示如何 mltable 使用 glob 模式(如通配符)来确保仅读取 Parquet 文件。

将此代码片段中的占位符 (<>) 替换为值:

import mltable

path = {
    'pattern': 'abfss://<filesystem>@<account>.dfs.core.windows.net/<folder>/*.parquet'
}

tbl = mltable.from_parquet_files(paths=[path])
df = tbl.to_pandas_dataframe()
df.head()

读取数据资产

本部分介绍了如何访问 Pandas 中的 Azure 机器学习数据资产。

表资产

如果以前在 Azure 机器学习中创建了一个表资产(一个mltable或一个 V1TabularDataset),则可以使用此代码将该表资产加载到 Pandas 中:

import mltable
from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential

ml_client = MLClient.from_config(credential=DefaultAzureCredential())
data_asset = ml_client.data.get(name="<name_of_asset>", version="<version>")

tbl = mltable.load(f'azureml:/{data_asset.id}')
df = tbl.to_pandas_dataframe()
df.head()

文件资产

如果已注册文件资产(例如 CSV 文件),可以使用以下代码将该资产读取到 Pandas 数据帧中:

import mltable
from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential

ml_client = MLClient.from_config(credential=DefaultAzureCredential())
data_asset = ml_client.data.get(name="<name_of_asset>", version="<version>")

path = {
    'file': data_asset.path
}

tbl = mltable.from_delimited_files(paths=[path])
df = tbl.to_pandas_dataframe()
df.head()

文件夹资产

如果已注册文件夹资产(uri_folder 或 V1 FileDataset),例如,包含 CSV 文件的文件夹,则可以使用此代码将该资产读取到 Pandas 数据帧中:

import mltable
from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential

ml_client = MLClient.from_config(credential=DefaultAzureCredential())
data_asset = ml_client.data.get(name="<name_of_asset>", version="<version>")

path = {
    'folder': data_asset.path
}

tbl = mltable.from_delimited_files(paths=[path])
df = tbl.to_pandas_dataframe()
df.head()

使用 Pandas 读取和处理大量数据

提示

Pandas 不设计用于处理大型数据集。 Pandas 只能处理能够装入计算实例内存的数据。

对于大型数据集,请使用 Azure 机器学习托管 Spark。 此服务提供 PySpark Pandas API

在纵向扩展到远程异步作业之前,你可能需要快速迭代大型数据集的较小子集。 mltable 提供通过 take_random_sample 方法获取大数据示例的功能:

import mltable

path = {
    'file': 'https://raw.githubusercontent.com/pandas-dev/pandas/main/doc/data/titanic.csv'
}

tbl = mltable.from_delimited_files(paths=[path])
# Take a random 30% sample of the data.
tbl = tbl.take_random_sample(probability=.3)
df = tbl.to_pandas_dataframe()
df.head()

还可以通过以下操作来获取大数据的子集:

使用 azcopy 实用工具下载数据

使用 azcopy 实用工具将数据下载到主机 SSD 的本地文件系统(本地计算机、云 VM、Azure 机器学习计算实例等)。 预安装在 Azure 机器学习计算实例上的 azcopy 实用工具将处理数据下载。 如果不使用 Azure 机器学习计算实例或 Data Science Virtual Machine (DSVM),则可能需要安装 azcopy。 有关详细信息,请参阅 azcopy

警告

不要将数据下载到 /home/azureuser/cloudfiles/code 计算实例上的位置。 此位置旨在存储笔记本和代码项目,而不是数据。 从此位置读取数据在训练期间会产生显著的性能损耗。 而是将数据存储在 home/azureuser 计算节点的本地 SSD 的位置。

打开终端并创建一个新目录,例如:

mkdir /home/azureuser/data

登录 azcopy

azcopy login

现在可以使用存储 URI 复制数据:

SOURCE=https://<account_name>.blob.core.windows.net/<container>/<path>
DEST=/home/azureuser/data
azcopy cp $SOURCE $DEST

后续步骤