昇思MindSpore进阶教程--单节点数据缓存(上)

大家好，我是刘明，明志科技创始人，华为昇思 MindSpore布道师。
技术上主攻前端开发、鸿蒙开发和AI算法研究。
努力为大家带来持续的技术分享，如果你也喜欢我的文章，就点个关注吧

正文开始

对于需要重复访问远程的数据集或需要重复从磁盘中读取数据集的情况，可以使用单节点缓存将数据集缓存于本地内存中，以加速数据集的读取。

缓存算子依赖于在当前节点启动的缓存服务器，缓存服务器作为守护进程独立于用户的训练脚本而存在，主要用于提供缓存数据的管理，支持包括存储、查找、读取以及发生缓存未命中时对于缓存数据的写入等操作。

若用户的内存空间不足以缓存所有数据集，则用户可以配置缓存算子使其将剩余数据缓存至磁盘。

目前，缓存服务只支持单节点缓存，即客户端和服务器均在同一台机器上。该服务支持以下两类使用场景：

缓存加载好的原始数据集

用户可以在数据集加载操作中使用缓存。这将把加载完成的数据存到缓存服务器中，后续若需相同数据则可直接从中读取，避免从磁盘中重复加载。
在这里插入图片描述

缓存经过数据增强处理后的数据

用户也可在map操作中使用缓存。这将允许直接缓存数据增强（如图像裁剪、缩放等）处理后的数据，避免数据增强操作重复进行，减少了不必要的计算量。
在这里插入图片描述

数据缓存流程

使用缓存服务前，需要安装MindSpore，并设置相关环境变量。

1. 启动缓存服务器

在使用单节点缓存服务之前，首先需要在命令行输入以下命令，启动缓存服务器：

!cache_admin --start

以上命令均可使用-h和-p参数来指定服务器，用户也可通过配置环境变量MS_CACHE_HOST和MS_CACHE_PORT来指定。若未指定则默认对ip为127.0.0.1且端口号为50052的服务器执行操作。

可通过ps -ef|grep cache_server命令来检查服务器是否已启动以及查询服务器参数。

也可通过cache_admin --server_info命令查看服务器的详细参数列表。

若要启用数据溢出功能，则用户在启动缓存服务器时必须使用-s参数对溢出路径进行设置，否则该功能默认关闭。

!cache_admin --server_info

2. 创建缓存会话

若缓存服务器中不存在缓存会话，则需要创建一个缓存会话，得到缓存会话id：

!cache_admin -g

通过cache_admin --list_sessions命令可以查看当前服务器中现存的所有缓存会话信息。

!cache_admin --list_sessions

3. 创建缓存实例

在Python训练脚本中使用DatasetCache 来定义一个名为test_cache的缓存实例，并把上一步中创建的缓存会话id传入session_id参数：

import os
import mindspore.dataset as ds# define a variable named `session_id` to receive the cache session ID created in the previous step
session_id = int(os.popen('cache_admin --list_sessions | tail -1 | awk -F " " \'{{print $1;}}\'').read())
test_cache = ds.DatasetCache(session_id=session_id, size=0, spilling=False)

在实际使用中，通常应当首先使用cache_admin -g命令从缓存服务器处获得一个缓存会话id并作为session_id的参数，防止发生缓存会话不存在而报错的情况。

设置size=0代表不限制缓存所使用的内存空间，缓存服务器会根据系统的内存资源状况，自动控制缓存服务器的内存空间占用，使其不超过系统总内存的80%。

用户也可以根据机器本身的空闲内存大小，给size参数设置一个合理的取值。注意，当用户自主设置size参数时，要先确认系统可用内存和待加载数据集大小，若cache_server的内存空间占用或待加载数据集空间占耗超过系统可用内存时，有可能导致机器宕机/重启、cache_server自动关闭、训练流程执行失败等问题。

若设置spilling=True，则当内存空间不足时，多余数据将写入磁盘中。因此，用户需确保所设置的磁盘路径具有写入权限以及足够的磁盘空间，以存储溢出至磁盘的缓存数据。注意，若启动服务器时未指定溢出路径，则在调用API时设置spilling=True将会导致报错。

若设置spilling=False，则缓存服务器在耗尽所设置的内存空间后将不再写入新的数据。

当使用不支持随机访问的数据集（如TFRecordDataset）进行数据加载并启用缓存服务时，需要保证整个数据集均存放于本地。在该场景下，若本地内存空间不足以存放所有数据，则必须启用溢出，将数据溢出至磁盘。

4. 插入缓存实例

当前缓存服务既支持对原始数据集的缓存，也可以用于缓存经过数据增强处理后的数据。下例分别展示了两种使用方式。

需要注意的是，两个例子均需要按照步骤3中的方法分别创建一个缓存实例，并在数据集加载或map操作中将所创建的test_cache作为cache参数分别传入。

下面样例中使用到CIFAR-10数据集。

from download import download
import os
import shutilurl = "https://mindspore-website.obs.cn-north-4.myhuaweicloud.com/notebook/datasets/cifar-10-binary.tar.gz"
path = download(url, "./datasets", kind="tar.gz", replace=True)test_path = "./datasets/cifar-10-batches-bin/test"
train_path = "./datasets/cifar-10-batches-bin/train"
os.makedirs(test_path, exist_ok=True)
os.makedirs(train_path, exist_ok=True)
if not os.path.exists(os.path.join(test_path, "test_batch.bin")):shutil.move("./datasets/cifar-10-batches-bin/test_batch.bin", test_path)
[shutil.move("./datasets/cifar-10-batches-bin/"+i, train_path) for i in os.listdir("./datasets/cifar-10-batches-bin/") if os.path.isfile("./datasets/cifar-10-batches-bin/"+i) and not i.endswith(".html") and not os.path.exists(os.path.join(train_path, i))]

解压后的数据集文件的目录结构如下：

./datasets/cifar-10-batches-bin
├── readme.html
├── test
│   └── test_batch.bin
└── train├── batches.meta.txt├── data_batch_1.bin├── data_batch_2.bin├── data_batch_3.bin├── data_batch_4.bin└── data_batch_5.bin

缓存原始数据集数据

缓存原始数据集，经过MindSpore系统加载后的数据。

dataset_dir = "./datasets/cifar-10-batches-bin/train"# apply cache to dataset
data = ds.Cifar10Dataset(dataset_dir=dataset_dir, num_samples=4, shuffle=False, num_parallel_workers=1, cache=test_cache)num_iter = 0
for item in data.create_dict_iterator(num_epochs=1):# in this example, each dictionary has a key "image"print("{} image shape: {}".format(num_iter, item["image"].shape))num_iter += 1

通过cache_admin --list_sessions命令可以查看当前会话有四条数据，说明数据缓存成功。

缓存经过增强后数据

缓存经过数据增强处理transforms后的数据。

import mindspore.dataset.vision as visiondataset_dir = "./datasets/cifar-10-batches-bin/train"# apply cache to dataset
data = ds.Cifar10Dataset(dataset_dir=dataset_dir, num_samples=5, shuffle=False, num_parallel_workers=1)# apply cache to map
rescale_op = vision.Rescale(1.0 / 255.0, -1.0)test_cache = ds.DatasetCache(session_id=session_id, size=0, spilling=False)data = data.map(input_columns=["image"], operations=rescale_op, cache=test_cache)num_iter = 0
for item in data.create_dict_iterator(num_epochs=1):# in this example, each dictionary has a keys "image"print("{} image shape: {}".format(num_iter, item["image"].shape))num_iter += 1

通过cache_admin --list_sessions命令可以查看当前会话有五条数据，说明数据缓存成功。

5. 销毁缓存会话

在训练结束后，可以选择将当前的缓存销毁并释放内存：

destroy_session = 'cache_admin --destroy_session' + str(session_id)

若选择不销毁缓存，则该缓存会话中的缓存数据将继续存在，用户下次启动训练脚本时可以继续使用该缓存。

6. 关闭缓存服务器

使用完毕后，可以通过以下命令关闭缓存服务器，该操作将销毁当前服务器中存在的所有缓存会话并释放内存。

!cache_admin --stop

以上命令将关闭端口50052的服务器。

若选择不关闭服务器，则服务器中已创建的缓存会话将保留，并供下次使用。下次训练时，用户可以新建缓存会话或重复使用已有缓存。