你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站,请访问 https://docs.azure.cn。
适用于:
Python SDK azure-ai-ml v2(当前)
机器学习项目通常从探索数据分析(EDA)和数据预处理(清理、特征工程)开始。 它还包括生成机器学习模型原型来验证假设。 此 原型制作 项目阶段具有高度交互性,它适合在 Jupyter 笔记本或 IDE 中使用 Python 交互式控制台进行开发。 在这篇文章中,你将学会如何:
- 像访问文件系统一样访问 Azure 机器学习数据存储 URI 中的数据。
- 使用
mltablePython 库将 Azure 机器学习数据资产具体化到 Pandas 中。 - 使用
azcopy工具通过显式下载将数据实体化。
Prerequisites
- Azure 机器学习工作区。 有关详细信息,请参阅 门户中或 Python SDK (v2)中的“管理 Azure 机器学习工作区”。
- Azure 机器学习数据存储。 有关详细信息,请参阅创建数据存储。
提示
本文中的指南介绍了交互式开发期间的数据访问。 它适用于可以运行 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:
- 在左侧菜单中选择 “数据” ,然后选择“ 数据存储 ”选项卡。
- 选择数据存储名称,然后选择“ 浏览”。
- 找到要读入 Pandas 的文件或文件夹,选择其旁边的省略号(...),然后选择 “复制 URI”。 然后,可以选择要复制到笔记本或脚本中的数据存储 URI。
还可以实例化 Azure 机器学习文件系统来处理类似文件系统的命令,例如ls、glob和existsopen。
-
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 进程文件。 此代码示例展示了如何使用 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 资源中的数据,则需要:
- 实例名称,格式为
adb-<number>.<number>.azuredatabricks.net. 可以从 Azure Databricks 工作区的 URL 获取此名称。 - 个人访问令牌。 有关个人访问令牌创建的详细信息,请参阅 使用 Azure Databricks 个人访问令牌(旧版)进行身份验证。
使用这些值,在计算实例上为 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