基于imagenet数据集的ResNet50模型训练示例

训练前准备

数据集获取

本训练示例以imagenet数据集为例，从imagenet官方网站http://www.image-net.org/获取数据集。

模型功能描述

Resnet50为一个深度残差网络，可用于对CIFAR-10和ImageNet的1000类数据集进行分类。

目录结构

主要文件目录结构如下所示(只列出部分涉及文件，更多文件请查看获取的ResNet原始网络脚本)：├── r1 // 原始模型目录

│ ├── resnet // resnet主目录

│ ├── __init__.py

│ ├── imagenet_main.py // 基于Imagenet数据集训练网络模型

│ ├── imagenet_preprocessing.py // Imagenet数据集数据预处理模块

│ ├── resnet_model.py // resnet模型文件

│ ├── resnet_run_loop.py // 数据输入处理与运行循环(训练、验证、测试)

│ ├── README.md // 项目介绍文件

│ ├── utils

│ │ ├── export.py // 数据接收函数，定义了导出的模型将会对何种格式的参数予以响应

├── utils

│ ├── flags

│ │ ├── core.py // 包含了参数定义的公共接口

│ ├── logs

│ │ ├── hooks_helper.py //自定义创建模型在测试/训练时的工具，比如每秒钟计算步数的功能、每N步或捕获CPU/GPU分析信息的功能等

│ │ ├── logger.py // 日志工具

│ ├── misc

│ │ ├── distribution_utils.py // 进行分布式运行模型的辅助函数

│ │ ├── model_helpers.py // 定义了一些能被模型调用的函数，比如控制模型是否停止

迁移说明

以下迁移过程，我们不借助迁移工具进行迁移，按照完全手工修改脚本的方式进行迁移，让大家更详细地了解当前脚本的所有迁移要点。

训练流程

Estimator简介

Estimator API属于TensorFlow的高阶API，在2018年发布的TensorFlow 1.10版本中引入，它可极大简化机器学习的编程过程。Estimator有很多优势，例如：对分布式的良好支持、简化了模型的创建工作、有利于模型开发者之间的代码分享等。

使用Estimator进行训练脚本开发的流程为：

表10-1 训练流程说明过程

描述

数据预处理

创建输入函数input_fn。

模型构建

构建模型函数model_fn。

运行配置

实例化Estimator，并传入Runconfig类对象作为运行参数。

执行训练

在Estimator上调用训练方法Estimator.train()，利用指定输入对模型进行固定步数的训练。

训练代码目录

目录结构

主要文件目录结构如下所示(只列出部分需要修改文件，更多文件请查看获取的ResNet原始网络脚本)：├── r1

│ ├── resnet // resnet主目录

│ ├── imagenet_main.py // 基于Imagenet数据集训练网络模型

│ ├── imagenet_preprocessing.py // Imagenet数据集数据预处理模块

│ ├── resnet_model.py // resnet模型文件

│ ├── resnet_run_loop.py // 数据输入处理与运行循环(训练、验证、测试)

├── utils

│ ├── flags

│ │ ├── _base.py //定义模型的通用参数并设置默认值

目录文件简介

表10-2 py文件作用及功能文件名称

简介

imagenet_main.py

包含imagenet数据集数据预处理、模型构建定义、模型运行的相关函数接口。其中数据预处理部分包含get_filenames()、parse_record()、input_fn()、get_synth_input_fn()，_parse_example_proto()函数，模型部分包含ImagenetModel类、imagenet_model_fn()、run_cifar()、define_cifar_flags()函数。

imagenet_preprocessing.py

imagenet图像数据预处理接口，训练过程中包括使用提供的边界框对训练图像进行采样、将图像裁剪到采样边界框、随机翻转图像，然后调整到目标输出大小(不保留纵横比)。评估过程中使用图像大小调整(保留纵横比)和中央剪裁。

resnet_model.py

ResNet模型的实现，包括辅助构建ResNet模型的函数以及ResNet block定义函数。

resnet_run_loop.py

模型运行文件，包括输入处理和运行循环两部分，输入处理包括对输入数据进行解码和格式转换，输出image和label，还根据是否是训练过程对数据的随机化、批次、预读取等细节做出了设定；运行循环部分包括构建Estimator，然后进行训练和验证过程。总体来看，是将模型放置在具体的环境中，实现数据与误差在模型中的流动，进而利用梯度下降法更新模型参数。

数据预处理

数据预处理流程与原始模型一致，部分位置经改造以适配昇腾910 AI处理器并提升计算性能，展示的示例代码包含改动位置。

定义输入函数input_fn

这里以imagenet数据集数据预处理为例，其数据预处理部分涉及到的适配昇腾910AI处理器改造的py文件及相关函数接口介绍如下：

表10-3 数据预处理API接口

简介

位置

input_fn()

输入函数，用于处理数据集用于Estimator训练，输出真实数据。

“/official/r1/resnet/imagenet_main.py”

resnet_main()

包含数据输入、运行配置、训练以及验证的主接口。

“/official/r1/resnet/resnet_run_loop.py”

在“official/r1/resnet/imagenet_main.py”文件中增加以下头文件：from hccl.manage.api import get_rank_size

from hccl.manage.api import get_rank_id

在数据读取时，获取芯片数量以及芯片id，用于支持数据并行。

代码位置：“official/r1/resnet/imagenet_main.py”的input_fn()函数(修改部分为字体加粗部分)：

definput_fn(is_training, data_dir, batch_size, num_epochs=1, dtype=tf.float32,

datasets_num_private_threads=None, parse_record_fn=parse_record,

input_context=None, drop_remainder=False, tf_data_experimental_slack=False):

"""提供训练和验证batches的函数。

参数解释:

is_training: 表示输入是否用于训练的布尔值。

data_dir: 包含输入数据集的文件路径。

batch_size: 每个batch的大小。

num_epochs: 数据集的重复数。

dtype: 图片/特征的数据类型。

datasets_num_private_threads: tf.data的专用线程数。

parse_record_fn: 解析tfrecords的入口函数。

input_context: 由'tf.distribute.Strategy'传入的'tf.distribute.InputContext'对象。

drop_remainder: 用于标示对于最后一个batch如果数据量达不到batch_size时保留还是抛弃。设置为True,则batch的维度固定。

tf_data_experimental_slack: 是否启用tf.data的'experimental_slack'选项。

Returns:

返回一个可用于迭代的数据集。

"""

# 获取文件路径

filenames = get_filenames(is_training, data_dir)

# 按第一个维度切分文件

dataset = tf.data.Dataset.from_tensor_slices(filenames)

ifinput_context:

# 获取芯片数量以及芯片id，用于支持数据并行

############## npu modify begin #############

dataset = dataset.shard(get_rank_size(),get_rank_id())

############## npu modify end ###############

# tf.compat.v1.logging.info(

# 'Sharding the dataset: input_pipeline_id=%d num_input_pipelines=%d' % (

# input_context.input_pipeline_id, input_context.num_input_pipelines))

# dataset = dataset.shard(input_context.num_input_pipelines,

# input_context.input_pipeline_id)

if is_training:

# 将文件顺序打乱

dataset = dataset.shuffle(buffer_size=_NUM_TRAIN_FILES)

# cycle_length = 10 并行读取并反序列化10个文件，CPU资源充足的场景下可适当增加该值。

dataset = dataset.interleave(

tf.data.TFRecordDataset,

cycle_length=10,

num_parallel_calls=tf.data.experimental.AUTOTUNE)

returnresnet_run_loop.process_record_dataset(

dataset=dataset,

is_training=is_training,

batch_size=batch_size,

shuffle_buffer=_SHUFFLE_BUFFER,

parse_record_fn=parse_record_fn,

num_epochs=num_epochs,

dtype=dtype,

datasets_num_private_threads=datasets_num_private_threads,

drop_remainder=drop_remainder,

tf_data_experimental_slack=tf_data_experimental_slack,

)

用于训练和测试的输入函数接口中，drop_remainder必须设置为True。

代码位置：“/official/r1/resnet/resnet_run_loop.py”中的resnet_main()函数(修改部分为input_fn_train()和input_fn_eval()子函数)：definput_fn_train(num_epochs, input_context=None):

############## npu modify begin #############

#使用dtype=tf.float16提高数据传输性能。

# 当前版本的drop_remainder只支持为True。

# 这里的batch_size指的是单卡的batch大小而不是全局batch大小。

returninput_function(

is_training=True,

data_dir=flags_obj.data_dir,

batch_size=flags_obj.batch_size,

num_epochs=num_epochs,

dtype=tf.float16,

input_context=input_context,

drop_remainder=True)

definput_fn_eval():

#使用dtype=tf.float16提高数据传输性能

# 当前版本的drop_remainder只支持为True

# 这里的batch_size指的是单卡的batch大小而不是全局batch大小

returninput_function(

is_training=False,

data_dir=flags_obj.data_dir,

batch_size=flags_obj.batch_size,

num_epochs=1,

dtype=tf.float16,

input_context=True,

drop_remainder=True)

############## npu modify end ###############

# 原代码中用于训练的输入函数接口和用于验证的输入函数接口。

# def input_fn_train(num_epochs, input_context=None):

# return input_function(

# is_training=True,

# data_dir=flags_obj.data_dir,

# batch_size=distribution_utils.per_replica_batch_size(

# flags_obj.batch_size, flags_core.get_num_gpus(flags_obj)),

# num_epochs=num_epochs,

# dtype=flags_core.get_tf_dtype(flags_obj),

# datasets_num_private_threads=flags_obj.datasets_num_private_threads,

# input_context=input_context)

#

# def input_fn_eval():

# return input_function(

# is_training=False,

# data_dir=flags_obj.data_dir,

# batch_size=distribution_utils.per_replica_batch_size(

# flags_obj.batch_size, flags_core.get_num_gpus(flags_obj)),

# num_epochs=1,

# dtype=flags_core.get_tf_dtype(flags_obj))

模型构建

模型构建与原始模型一致，部分位置经过适应性改造以提升计算性能，展示的示例代码包含改动位置。

定义模型函数

以根据imagenet构建的模型函数为例，其相关函数接口如下：

表10-4 模型函数相关接口类和接口

简介

位置

imagenet_model_fn()

基于imagenet构建的模型函数。

“/official/r1/resnet/imagenet_main.py”

learning_rate_with_decay()

建立学习率函数，当全局步数小于设定步数时，学习率线性增加，当超过设定步数时，学习率分阶段下降。

“/official/r1/resnet/resnet_run_loop.py”

resnet_model_fn()

用于构建EstimatorSpec，该类定义了由Estimator运行的模型。

“/official/r1/resnet/resnet_run_loop.py”

ImagenetModel()

ImagenetModel继承自resnet_model模块下的Model，指定了适用于imagenet的ResNet模型的网络规模、版本、分类数、卷积参数和池化参数等。

“/official/r1/resnet/imagenet_main.py”

__call__()

添加操作以对输入图片进行分类，包括：1、为了加速GPU运算，将输入由NHWC转换成NCHW；2、首次卷积运算；3、根据ResNet版本判断是否要做batch norm；4、首次pooling；5、堆叠block；6、计算输入图像的平均值；7、全连接层。

“/official/r1/resnet/resnet_model.py”

性能提升在“/official/r1/resnet/resnet_run_loop.py”增加以下头文件：from npu_bridge.hccl import hccl_ops

检查输入特征/图像数据类型。

代码位置：“official/r1/resnet/resnet_run_loop.py”的resnet_model_fn()函数(修改部分为字体加粗部分)：############# npu modify begin #############

# 检查输入特征/图像是否与用于计算的数据类型一致。

iffeatures.dtype != dtype:

# 将特征的数据类型改成与dtype一致。

features = tf.cast(features, dtype)

############## npu modify end ###############

# 原代码中数据类型修改如下：

# assert features.dtype == dtype

计算accuracy时labels使用float32类型以提升精度。

代码位置：“official/r1/resnet/resnet_run_loop.py”的resnet_model_fn()函数(修改部分为字体加粗部分)：############## npu modify begin #############

# labels使用float32类型来提升精度。

accuracy = tf.compat.v1.metrics.accuracy(tf.cast(labels, tf.float32), predictions['classes'])

############## npu modify end ###############

# 原代码中计算accuracy如下：

# accuracy = tf.compat.v1.metrics.accuracy(labels, predictions['classes'])

accuracy_top_5 = tf.compat.v1.metrics.mean(

tf.nn.in_top_k(predictions=logits, targets=labels, k=5, name='top_5_op'))

############## npu modify begin #############

# 用于分布式训练时的accuracy计算。

rank_size = int(os.getenv('RANK_SIZE'))

newaccuracy = (hccl_ops.allreduce(accuracy[0], "sum") / rank_size, accuracy[1])

newaccuracy_top_5 = (hccl_ops.allreduce(accuracy_top_5[0], "sum") / rank_size, accuracy_top_5[1])

metrics = {'accuracy': newaccuracy,

'accuracy_top_5': newaccuracy_top_5}

############## npu modify end #############

# 原代码中的metrics表示如下：

# metrics = {'accuracy': accuracy,

# 'accuracy_top_5': accuracy_top_5}

使用max_pool_with_argmax算子替代max_pooling2d算子，以获得更好的计算性能。

代码位置：“official/r1/resnet/resnet_model.py”的__call__()函数(修改部分为字体加粗部分)：# 是否进行第一次池化。

if self.first_pool_size:

############## npu modify begin #############

# 使用max_pool_with_argmax代替max_pooling2d能获得更好的表现。

inputs,argmax = tf.compat.v1.nn.max_pool_with_argmax(

input=inputs, ksize=(1,self.first_pool_size,self.first_pool_size,1),

strides=(1,self.first_pool_stride,self.first_pool_stride,1), padding='SAME',

data_format='NCHW' if self.data_format == 'channels_first' else 'NHWC')

############## npu modify end ###############

# 原代码使用max_pooling2d()接口进行池化

# inputs = tf.compat.v1.layers.max_pooling2d(

# inputs=inputs, pool_size=self.first_pool_size,

# strides=self.first_pool_stride, padding='SAME',

# data_format=self.data_format)

inputs = tf.identity(inputs, 'initial_max_pool')

分布式训练配置在“official/r1/resnet/resnet_run_loop.py”文件中增加以下头文件：from npu_bridge.estimator.npu.npu_optimizer import NPUDistributedOptimizer

添加分布式训练优化器NPUDistributedOptimizer，用于分布式训练。

代码位置：“official/r1/resnet/resnet_run_loop.py”的resnet_model_fn()函数(修改部分为字体加粗部分)：

if flags.FLAGS.enable_lars:

optimizer = tf.contrib.opt.LARSOptimizer(

learning_rate,

momentum=momentum,

weight_decay=weight_decay,

skip_list=['batch_normalization', 'bias'])

else:

optimizer = tf.compat.v1.train.MomentumOptimizer(

learning_rate=learning_rate,

momentum=momentum

)

############## npu modify begin #############

# 使用分布式训练优化器封装单机优化器，用于支持分布式训练。

# 在原代码中添加如下代码。

optimizer = NPUDistributedOptimizer(optimizer)

############## npu modify end ###############

fp16_implementation = getattr(flags.FLAGS, 'fp16_implementation', None)

if fp16_implementation == 'graph_rewrite':

optimizer = (

tf.compat.v1.train.experimental.enable_mixed_precision_graph_rewrite(

optimizer, loss_scale=loss_scale))

运行配置

运行配置

模型配置函数主要在resnet_main()中，其函数介绍如下：

表10-5 模型配置相关函数接口接口

简介

位置

resnet_main()

包含运行配置、训练以及验证的主要函数。

“/official/r1/resnet/resnet_run_loop.py”

在“/official/r1/resnet/resnet_run_loop.py”增加以下头文件：from npu_bridge.estimator.npu.npu_config import NPURunConfig

from npu_bridge.estimator.npu.npu_estimator import NPUEstimator

通过NPURunconfig替代Runconfig来配置运行参数。

代码位置：“official/r1/resnet/resnet_run_loop.py”的resnet_main()函数：############## npu modify begin #############

# 使用NPURunconfig替换Runconfig，适配昇腾AI处理器，每115200步保存一次checkpoint，每10000次保存一次summary，

# 对数据进行预处理，使用混合精度模式提升训练速度。

run_config = NPURunConfig(

model_dir=flags_obj.model_dir,

session_config=session_config,

save_checkpoints_steps=115200,

enable_data_pre_proc=True,

iterations_per_loop=100,

# enable_auto_mix_precision=True,

# 精度模式设置为混合精度模式。

precision_mode='allow_mix_precision',

hcom_parallel=True

)

############## npu modify end ###############

# 原代码中运行参数配置如下：

# run_config = tf.estimator.RunConfig(

# train_distribute=distribution_strategy,

# session_config=session_config,

# save_checkpoints_secs=60 * 60 * 24,

# save_checkpoints_steps=None)

关于设置混合精度模式：precision_mode='allow_mix_precision'，具体可参考混合精度。

创建NPUEstimator，使用NPUEstimator接口代替tf.estimator.Estimator。

代码位置：“/official/r1/resnet/resnet_run_loop.py”的resnet_main()函数(修改部分如下)：

# 使用`NPUEstimator`接口代替tf.estimator.Estimator

classifier = NPUEstimator(

model_fn=model_function, model_dir=flags_obj.model_dir, config=run_config,

params={

'resnet_size': int(flags_obj.resnet_size),

'data_format': flags_obj.data_format,

'batch_size': flags_obj.batch_size,

'resnet_version': int(flags_obj.resnet_version),

'loss_scale': flags_core.get_loss_scale(flags_obj,

default_for_fp16=128),

'dtype': flags_core.get_tf_dtype(flags_obj),

'fine_tune': flags_obj.fine_tune,

'num_workers': num_workers,

'num_gpus': flags_core.get_num_gpus(flags_obj),

})

# 原代码中创建Estimator如下：

# classifier = tf.estimator.Estimator(

# model_fn=model_function, model_dir=flags_obj.model_dir, config=run_config,

# warm_start_from=warm_start_settings, params={

# 'resnet_size': int(flags_obj.resnet_size),

# 'data_format': flags_obj.data_format,

# 'batch_size': flags_obj.batch_size,

# 'resnet_version': int(flags_obj.resnet_version),

# 'loss_scale': flags_core.get_loss_scale(flags_obj,

# default_for_fp16=128),

# 'dtype': flags_core.get_tf_dtype(flags_obj),

# 'fine_tune': flags_obj.fine_tune,

# 'num_workers': num_workers,

# })

执行训练

训练模块

训练模块函数接口如下：

表10-6 训练模块函数接口接口

简介

位置

main()

主函数入口，包含训练之前集合通信初始化，执行模型训练。

“/official/r1/resnet/imagenet_main.py”

run_imagenet()

模型训练入口，包含输入函数选择及返回训练结果。

“/official/r1/resnet/imagenet_main.py”

resnet_main()

包含运行配置、训练以及验证的主要函数。

“/official/r1/resnet/resnet_run_loop.py”

分布式训练在“official/r1/resnet/resnet_run_loop.py”增加以下头文件：from npu_bridge.estimator import npu_ops

from tensorflow.core.protobuf import rewriter_config_pb2

训练之前集合通信初始化。

代码位置：“official/r1/resnet/imagenet_main.py”的main()函数(修改部分如下)：def main():

############## npu modify begin #############

# 初始化NPU，调用HCCL接口。

# 在原代码中添加如下内容：

init_sess, npu_init = resnet_run_loop.init_npu()

init_sess.run(npu_init)

############## npu modify end ###############

with logger.benchmark_context(flags.FLAGS):

run_imagenet(flags.FLAGS)

集合通信初始化函数定义。

代码位置：“official/r1/resnet/resnet_run_loop.py”中添加init_npu()函数接口：def resnet_main(flags_obj, model_function, input_function, dataset_name, shape=None):...

############## npu modify begin #############

# 添加如下代码

def init_npu():

"""此接口作用是手动初始化NPU。

返回值：

`init_sess` npu init session config.

`npu_init` npu init ops.

"""

npu_init = npu_ops.initialize_system()

config = tf.ConfigProto()

config.graph_options.rewrite_options.remapping = rewriter_config_pb2.RewriterConfig.OFF

custom_op = config.graph_options.rewrite_options.custom_optimizers.add()

custom_op.name = "NpuOptimizer"

# custom_op.parameter_map["precision_mode"].b = True

custom_op.parameter_map["precision_mode"].s = tf.compat.as_bytes("allow_mix_precision")

custom_op.parameter_map["use_off_line"].b = True

init_sess = tf.Session(config=config)

return init_sess, npu_init

############## npu modify end ###############

单次训练/验证结束后释放设备资源。

代码位置：“official/r1/resnet/resnet_run_loop.py”中的resnet_main()函数中(修改部分如图中加粗部分)。for cycle_index, num_train_epochs in enumerate(schedule):

tf.compat.v1.logging.info('Starting cycle: %d/%d', cycle_index,

int(n_loops))

if num_train_epochs:

# Since we are calling classifier.train immediately in each loop, the

# value of num_train_epochs in the lambda function will not be changed

# before it is used. So it is safe to ignore the pylint error here

# pylint: disable=cell-var-from-loop

classifier.train(

input_fn=lambda input_context=None: input_fn_train(

num_train_epochs, input_context=input_context),

hooks=train_hooks,

max_steps=flags_obj.max_train_steps)

############## npu modify begin #############

# 单次训练结束时，释放npu资源，在下一次进程开始之前如果要用到hccl接口需要重新初始化。

# 在原代码中添加如下内容:

init_sess, npu_init = init_npu()

npu_shutdown = npu_ops.shutdown_system()

init_sess.run(npu_shutdown)

init_sess.run(npu_init)

############## npu modify end ###############

tf.compat.v1.logging.info('Starting to evaluate.')

eval_results = classifier.evaluate(input_fn=input_fn_eval,

steps=flags_obj.max_train_steps)

benchmark_logger.log_evaluation_result(eval_results)

if model_helpers.past_stop_threshold(

flags_obj.stop_threshold, eval_results['accuracy']):

break

############## npu modify begin #############

# 单次训练结束时，释放npu资源，在下一次进程开始之前如果要用到hccl接口需要重新初始化。

# 在原代码中添加如下内容:

init_sess, npu_init = init_npu()

npu_shutdown = npu_ops.shutdown_system()

init_sess.run(npu_shutdown)

init_sess.run(npu_init)

############## npu modify end ###############

所有训练/验证结束后释放设备资源。

在所有训练/验证结束后通过npu_ops.shutdown_system接口释放设备资源。

代码位置：“official/r1/resnet/resnet_run_loop.py”中的resnet_main()函数中(修改部分如下)：if flags_obj.export_dir is not None:

# Exports a saved model for the given classifier.

export_dtype = flags_core.get_tf_dtype(flags_obj)

if flags_obj.image_bytes_as_serving_input:

input_receiver_fn = functools.partial(

image_bytes_serving_input_fn, shape, dtype=export_dtype)

else:

input_receiver_fn = export.build_tensor_serving_input_receiver_fn(

shape, batch_size=flags_obj.batch_size, dtype=export_dtype)

classifier.export_savedmodel(flags_obj.export_dir, input_receiver_fn,

strip_default_attrs=True)

############## npu modify begin #############

# 在所有训练/验证结束后通过npu_ops.shutdown_system接口释放设备资源。在原代码中添加如下内容：

npu_shutdown = npu_ops.shutdown_system()

init_sess.run(npu_shutdown)

############## npu modify end ###############

stats = {}

stats['eval_results'] = eval_results

stats['train_hooks'] = train_hooks

return stats

loss scale设定

设定loss scale默认值。

代码位置：“official/r1/resnet/imagenet_main.py”的define_imagenet_flags()函数(修改部分如下)：def define_imagenet_flags():

resnet_run_loop.define_resnet_flags(

resnet_size_choices=['18', '34', '50', '101', '152', '200'],

dynamic_loss_scale=True,

fp16_implementation=True)

flags.adopt_module_key_flags(resnet_run_loop)

flags_core.set_defaults(train_epochs=90)

############## npu modify begin #############

# 昇腾AI处理器默认支持混合精度训练，loss_scale设置过大可能导致梯度爆炸，设置过小可能会导致梯度消失，

# 设置以下经验值能避免以上问题。

flags_core.set_defaults(loss_scale='512')

############## npu modify end ###############

脚本执行

准备数据集

准备数据集并上传到运行环境的目录下，例如：/home/data/resnet50/imagenet。

准备ranktable文件

ranktable文件样例和文件说明请参考芯片资源信息配置文件参考。

配置环境变量

请参考在裸机环境执行训练配置环境变量。

执行命令

python3 /home/official/r1/resnet/imagenet_main.py --batch_size=32 --hooks=ExamplesPerSecondHook --data_dir=/home/data/resnet50/imagenet