【2.4.1.5】tensorflow多GPU的实现-5
欢迎来到本课程的第三个实验。到目前为止，您已经了解了随机梯度下降以及批量大小对训练的影响。您还学习了如何将训练从单个GPU扩展到多个GPU，您应该已经看到了相当大的加速，我们将对其进行更详细的测量。
您已经看到，随着我们将训练扩展到多个GPU上，我们有效地增加了批量大小。根据我们在实验中进行的测试，我们知道增加批量大小可以提高吞吐量，但也可能影响网络有效训练的能力。在本实验中，您将探索在扩展到多个GPU时训练如何受到影响。我们将从测量多个GPU对吞吐量的影响开始，在本例中以每秒处理多少图像为指标。然后，您将继续学习可帮助您在扩展训练时保持精度的工具和技术。
一、性能的扩展：图像数/秒
在第一个实验中，您评估了图像数/每秒作为训练性能的度量。我们看到吞吐量随着批量大小的增加而增加，但仅能增加到一定程度。随着批量的增加直至GPU内存饱和，我们看到了收益递减。既然我们正在将训练扩展到多个GPU上，我们的吞吐量有理由再次增加。您认为训练中使用的GPU数量会如何影响吞吐量的增加呢？
像在第一个实验中的那样，我们有一个回调函数，它在每个训练周期之后以图像数/每秒为单位测量吞吐量。请在 一个GPU上 运行下面的代码单元，并在其下面的代码单元中记录吞吐量。您可能从实验1回忆起，应该忽略第一个训练周期的吞吐量，因为我们需要付出一次性的开销，因此请从第二个或第三个周期中选择一个吞吐量。
!mpirun -np FIXME python fashion_mnist.py --epochs 3 --batch-size 128
接下来，在所有的GPU上执行训练。 您可能会注意到吞吐量会在各个训练周期之间波动。请选择看起来最接近平均值的吞吐量值，或者计算几个周期的平均值，切记不要包括第一个周期的值。
num_gpus = FIXME
!mpirun -np $num_gpus python fashion_mnist.py --epochs 5 --batch-size 128
在下面的单元格中记录您获得的值，然后执行它以查看加速比。思考一下您得到的数字，然后通过单击三个点以显示下面的方框。
one_gpu_throughput = FIXME
multi_gpu_throughput = FIXME
print("Multi GPU speedup factor: {}".format(multi_gpu_throughput/one_gpu_throughput))
您应该看到吞吐量几乎以线性速率增加。与在实验1中看到的增加批量大小时获得的增加比例相比，这是一个更令人鼓舞的结果。可以肯定的是，每个GPU都会大大加快训练过程。
我们并未得到完美的线性增加，其中的一个重要原因是由于更新权重时GPU之间的通信，也可能有其它因素，例如在权重平均之前等待较慢的GPU完成处理。但是，这个结果还是不错的。
深度学习的大规模训练通常以线性增加的理想情况为基准（N个GPU应该比一个GPU快N倍）。 Horovod和NCCL库在保持高吞吐量方面做得很好，但是他们的性能与所使用的硬件有着千丝万缕的联系。高带宽和低延迟的要求导致了NVLink互连的开发，它是本课程所使用的服务器用来互连一个节点上的多个GPU的方法。 NVIDIA DGX-2通过NVSwitch将这种互连又推进一步，该互连结构可以300GB/s的峰值双向带宽连接多达16个GPU。
当您扩展到更多GPU时就需要进行跨多节点的训练，而这还需要考虑更多的硬件因素才能获得有效的性能扩展。
二、比较一段时间内的验证准确性
与第一个实验一样，吞吐量只是衡量性能的一种方法。评估网络的训练性能，尤其是其有效进行预测的能力，是非常重要的。为了清晰地显示不同训练之间的对比，我们将绘制网络的验证准确性随时间的变化图。
为此，您将保存验证准确性以及每个训练周期之后的总消耗时间。每个训练都将创建一个CSV文件，该文件将用于绘制各个训练过程。我们为您提供了一个脚本，以绘制下面的训练过程。
实现回调函数
首先，让我们复制训练脚本，以备您需要参考时使用。
!cp fashion_mnist.py fashion_mnist_original.py
接下来，查看callbacks/save_training_data.py。 请按照以下步骤在fashion_mnist.py中实现回调函数。您需要为回调函数传递CSV文件的名称，该名称将由您在训练中使用的超参数组成。如果您遇到麻烦并需要帮助，可以在solutions/save_training_data.py中找到解决方案。
执行训练会话
在实现了回调函数之后，请尝试在多个GPU上进行训练。我们建议在一定时间（100-200秒）后停止。您可以通过单击顶部栏中的停止按钮（或菜单中的Kernel > Interrupt）来执行此操作。您也可以像在实验1中一样使用–target-accuracy和–patience参数来停止训练。即使您提早结束训练，CSV文件也将正确保存。
# Add the training_data folder
!mkdir training_data
!mpirun -np $num_gpus python fashion_mnist.py --batch-size 32
绘制数据
如果回调函数工作正常，则在training_data文件夹中应该包含一个CSV文件。您随时可在左侧窗格中的文件浏览器中导航至该文件，并在那里打开文件。 然后在下面的单元格中查看其中的代码。它将绘制在training_data文件夹中的所有CSV文件包含的数据，并使用图例说明中的名称来标识每次训练。了解代码后，请执行该单元格以绘制训练图。
%matplotlib widget
import os
import numpy as np
import matplotlib.pyplot as plt
# By default we skip the first row, which contains the headers
# By skipping 2 rows, you can disregard the first data-point (0,0) to get a closer look
def plot_trainings(skiprows=1):
plt.close()
for filename in os.listdir("training_data"):
if filename == ".ipynb_checkpoints": continue
x, y = np.loadtxt("training_data/" + filename, delimiter=',', unpack=True, skiprows=skiprows)
plt.plot(x,y, label=filename.split('.csv')[0])
plt.xlabel('Time (s)')
plt.ylabel('Validation Accuracy')
plt.title('Training Comparison')
plt.legend()
plt.show()
plot_trainings()
清理数据
在您进行本实验中的练习时，您可能会发现该图显得很混乱，并且很难将一个训练与其他训练进行比较。发生这种情况时，建议您进入training_data文件夹，并根据需要删除文件。您可以使用以下命令清除目录中的所有内容：
# Only run this if you want to delete all training data
!rm -rf training_data/*
另请注意，如果您使用所有相同的超参数第二次运行训练（因此对CSV文件使用相同的名称），它将覆盖原来的训练文件。
三、调整优化器：增加动量
到目前为止，我们已经通过更改批量大小以及将训练分布到多个GPU上来修改我们的训练。我们稍后将在实验中探索其它超参数，例如学习率。在本节中，我们将探索改善训练过程的另一个关键工具：选择和调整我们使用的优化器。
您可能从实验1中回忆起，优化器会更新网络的权重，以使损失函数降到最低。到目前为止，在我们使用的随机梯度下降优化器中，权重的更新是基于其相对于小批量损失函数的梯度进行更新。换句话说，我们确定改变权重将如何影响损失，并朝最小化小批量下的损失的方向移动一小步。通过在每次反向传播中采取这些步骤，我们逐渐朝着全局最小值的方向发展，减少了损失，并提高了准确性。
尽管此过程运行良好，但还可以对其进行改进。SGD有一个缺点是，如果网络的损失接近局部最小值或鞍点，则梯度可能会很小，并且该区域的训练速度会变慢。使用小的批量所引入的梯度噪声可能有助于模型找到解决方法，但这可能需要一些时间。
另外，算法可能还会在某些区域沿相同方向继续执行步骤。一个更优越的算法是，它在这些区域里帮助我们采用更大的步，使训练更快地朝着全局最小值迈进。
解决这些问题的一个好办法是使用动量。向我们的优化器里添加动量因素可以使优化过程保留对最后几步的记忆，而不是每次都只迈一小步。一个贴切的比喻是，它像一个滚下山坡的球，它将积累并保持动量。
具体实现
在这种情况下，实现动量非常简单，因为我们不必更改优化程序的代码。 SGD算法通常会接受动量参数，例如Keras实现中的情况。
步骤1：更改参数解析器以接受动量参数：
parser.add_argument('--momentum', type=float, default=0.9,
help='SGD momentum')
步骤2：将动量传递给优化器
opt = tf.keras.optimizers.SGD(lr=args.base_lr, momentum=args.momentum)
步骤3：更新CSV文件名以包含动量超参数
data_filepath = "training_data/{}ranks-{}bs-{}lr-{}m.csv".format(hvd.size(), args.batch_size, args.base_lr, args.momentum)
如果您需要任何帮助，可以在solutions/add_momenum.py中找到解决方案。
用带动量的优化器进行训练 再次运行与以前相同时间的训练，或者使用–target-accuracy在达到一定的精度时停止训练。训练完成后，执行绘图脚本以查看有动量和无动量的训练之间的差异。
!mpirun -np $num_gpus python fashion_mnist.py --batch-size 32 --momentum .9
plot_trainings()
脚本
fashion_mnist.py：
import argparse
import tensorflow as tf
from tensorflow.keras import backend as K
from tensorflow.keras.preprocessing import image
from tensorflow.keras.datasets import fashion_mnist
from tensorflow.keras.layers import Input, Conv2D, BatchNormalization, Dense, \
Add, Activation, Dropout, MaxPooling2D, GlobalAveragePooling2D
import numpy as np
import os
from time import time
import horovod.tensorflow.keras as hvd
# Initialize Horovod
hvd.init()
# Pin to a GPU
gpus = tf.config.experimental.list_physical_devices('GPU')
if gpus:
tf.config.experimental.set_memory_growth(gpus[hvd.local_rank()], True)
tf.config.experimental.set_visible_devices(gpus[hvd.local_rank()], 'GPU')
# Parse input arguments
parser = argparse.ArgumentParser(description='Fashion MNIST Example',
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument('--log-dir', default='./logs',
help='tensorboard log directory')
parser.add_argument('--batch-size', type=int, default=32,
help='input batch size for training')
parser.add_argument('--val-batch-size', type=int, default=32,
help='input batch size for validation')
parser.add_argument('--epochs', type=int, default=40,
help='number of epochs to train')
parser.add_argument('--base-lr', type=float, default=0.01,
help='learning rate for a single GPU')
parser.add_argument('--wd', type=float, default=0.000005,
help='weight decay')
parser.add_argument('--target-accuracy', type=float, default=.85,
help='Target accuracy to stop training')
parser.add_argument('--patience', type=float, default=2,
help='Number of epochs that meet target before stopping')
args = parser.parse_args()
# Define a function for a simple learning rate decay over time
def lr_schedule(epoch):
if epoch < 15:
return args.base_lr
if epoch < 25:
return 1e-1 * args.base_lr
if epoch < 35:
return 1e-2 * args.base_lr
return 1e-3 * args.base_lr
# Define the function that creates the model
def cbr(x, conv_size):
channel_axis = 1 if K.image_data_format() == 'channels_first' else -1
x = Conv2D(conv_size, (3,3), padding='same')(x)
x = BatchNormalization(axis=channel_axis)(x)
x = Activation('relu')(x)
return x
def conv_block(x, conv_size, scale_input = False):
x_0 = x
if scale_input:
x_0 = Conv2D(conv_size, (1, 1), activation='linear', padding='same')(x_0)
x = cbr(x, conv_size)
x = Dropout(0.01)(x)
x = cbr(x, conv_size)
x = Add()([x_0, x])
return x
def create_model():
# Implementation of WideResNet (depth = 16, width = 10) based on keras_contrib
# https://github.com/keras-team/keras-contrib/blob/master/keras_contrib/applications/wide_resnet.py
inputs = Input(shape=(28, 28, 1))
x = cbr(inputs, 16)
x = conv_block(x, 160, True)
x = conv_block(x, 160)
x = MaxPooling2D((2, 2))(x)
x = conv_block(x, 320, True)
x = conv_block(x, 320)
x = MaxPooling2D((2, 2))(x)
x = conv_block(x, 640, True)
x = conv_block(x, 640)
x = GlobalAveragePooling2D()(x)
outputs = Dense(num_classes, activation='softmax')(x)
model = tf.keras.models.Model(inputs, outputs)
opt = tf.keras.optimizers.SGD(lr=args.base_lr)
# Wrap the optimizer in a Horovod distributed optimizer
opt = hvd.DistributedOptimizer(opt)
model.compile(loss=tf.keras.losses.categorical_crossentropy,
optimizer=opt,
metrics=['accuracy'])
return model
if hvd.rank() == 0:
verbose = 1
else:
verbose = 0
# Input image dimensions
img_rows, img_cols = 28, 28
num_classes = 10
# Load Fashion MNIST data.
(x_train, y_train), (x_test, y_test) = fashion_mnist.load_data()
if K.image_data_format() == 'channels_first':
x_train = x_train.reshape(x_train.shape[0], 1, img_rows, img_cols)
x_test = x_test.reshape(x_test.shape[0], 1, img_rows, img_cols)
input_shape = (1, img_rows, img_cols)
else:
x_train = x_train.reshape(x_train.shape[0], img_rows, img_cols, 1)
x_test = x_test.reshape(x_test.shape[0], img_rows, img_cols, 1)
input_shape = (img_rows, img_cols, 1)
# Convert class vectors to binary class matrices
y_train = tf.keras.utils.to_categorical(y_train, num_classes)
y_test = tf.keras.utils.to_categorical(y_test, num_classes)
# Training data iterator.
train_gen = image.ImageDataGenerator(featurewise_center=True, featurewise_std_normalization=True,
horizontal_flip=True, width_shift_range=0.2, height_shift_range=0.2)
train_gen.fit(x_train)
train_iter = train_gen.flow(x_train, y_train, batch_size=args.batch_size)
# Validation data iterator.
test_gen = image.ImageDataGenerator(featurewise_center=True, featurewise_std_normalization=True)
test_gen.mean = train_gen.mean
test_gen.std = train_gen.std
test_iter = test_gen.flow(x_test, y_test, batch_size=args.val_batch_size)
callbacks = []
callbacks.append(tf.keras.callbacks.LearningRateScheduler(lr_schedule))
# Broadcast initial variable states from the first worker to all others.
callbacks.append(hvd.callbacks.BroadcastGlobalVariablesCallback(0))
# Average the metrics among workers at the end of every epoch.
callbacks.append(hvd.callbacks.MetricAverageCallback())
class PrintThroughput(tf.keras.callbacks.Callback):
def __init__(self, total_images=0):
self.total_images = total_images
def on_epoch_begin(self, epoch, logs=None):
self.epoch_start_time = time()
def on_epoch_end(self, epoch, logs={}):
epoch_time = time() - self.epoch_start_time
images_per_sec = round(self.total_images / epoch_time, 2)
print('Images/sec: {}'.format(images_per_sec))
if verbose:
callbacks.append(PrintThroughput(total_images=len(y_train)))
class StopAtAccuracy(tf.keras.callbacks.Callback):
def __init__(self, target=0.85, patience=2, verbose=0):
self.target = target
self.patience = patience
self.verbose = verbose
self.stopped_epoch = 0
self.met_target = 0
def on_epoch_end(self, epoch, logs=None):
if logs.get('val_accuracy') > self.target:
self.met_target += 1
else:
self.met_target = 0
if self.met_target >= self.patience:
self.stopped_epoch = epoch
self.model.stop_training = True
def on_train_end(self, logs=None):
if self.stopped_epoch > 0 and self.verbose == 1:
print('Early stopping after epoch {}'.format(self.stopped_epoch + 1))
callbacks.append(StopAtAccuracy(target=args.target_accuracy, patience=args.patience, verbose=verbose))
class PrintTotalTime(tf.keras.callbacks.Callback):
def on_train_begin(self, logs=None):
self.start_time = time()
def on_epoch_end(self, epoch, logs=None):
total_time = round(time() - self.start_time, 2)
print("Cumulative training time after epoch {}: {}".format(epoch + 1, total_time))
def on_train_end(self, logs=None):
total_time = round(time() - self.start_time, 2)
print("Cumulative training time: {}".format(total_time))
if verbose:
callbacks.append(PrintTotalTime())
# Create the model.
model = create_model()
# Train the model.
model.fit(train_iter,
steps_per_epoch=len(train_iter) // hvd.size(),
callbacks=callbacks,
epochs=args.epochs,
verbose=verbose,
workers=4,
initial_epoch=0,
validation_data=test_iter,
validation_steps=3 * len(test_iter) // hvd.size())
# Evaluate the model on the full data set.
score = model.evaluate(test_iter, steps=len(test_iter), workers=4, verbose=verbose)
if verbose:
print('Test loss:', score[0])
print('Test accuracy:', score[1])
save_training_data.py：
import argparse
import tensorflow as tf
from tensorflow.keras import backend as K
from tensorflow.keras.preprocessing import image
from tensorflow.keras.datasets import fashion_mnist
from tensorflow.keras.layers import Input, Conv2D, BatchNormalization, Dense, \
Add, Activation, Dropout, MaxPooling2D, GlobalAveragePooling2D
import numpy as np
import os
from time import time
import horovod.tensorflow.keras as hvd
import csv
# Initialize Horovod
hvd.init()
# Pin to a GPU
gpus = tf.config.experimental.list_physical_devices('GPU')
if gpus:
tf.config.experimental.set_memory_growth(gpus[hvd.local_rank()], True)
tf.config.experimental.set_visible_devices(gpus[hvd.local_rank()], 'GPU')
# Parse input arguments
parser = argparse.ArgumentParser(description='Fashion MNIST Example',
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument('--log-dir', default='./logs',
help='tensorboard log directory')
parser.add_argument('--batch-size', type=int, default=32,
help='input batch size for training')
parser.add_argument('--val-batch-size', type=int, default=32,
help='input batch size for validation')
parser.add_argument('--epochs', type=int, default=40,
help='number of epochs to train')
parser.add_argument('--base-lr', type=float, default=0.01,
help='learning rate for a single GPU')
parser.add_argument('--wd', type=float, default=0.000005,
help='weight decay')
parser.add_argument('--target-accuracy', type=float, default=.85,
help='Target accuracy to stop training')
parser.add_argument('--patience', type=float, default=2,
help='Number of epochs that meet target before stopping')
args = parser.parse_args()
# Define a function for a simple learning rate decay over time
def lr_schedule(epoch):
if epoch < 15:
return args.base_lr
if epoch < 25:
return 1e-1 * args.base_lr
if epoch < 35:
return 1e-2 * args.base_lr
return 1e-3 * args.base_lr
# Define the function that creates the model
def cbr(x, conv_size):
channel_axis = 1 if K.image_data_format() == 'channels_first' else -1
x = Conv2D(conv_size, (3,3), padding='same')(x)
x = BatchNormalization(axis=channel_axis)(x)
x = Activation('relu')(x)
return x
def conv_block(x, conv_size, scale_input = False):
x_0 = x
if scale_input:
x_0 = Conv2D(conv_size, (1, 1), activation='linear', padding='same')(x_0)
x = cbr(x, conv_size)
x = Dropout(0.01)(x)
x = cbr(x, conv_size)
x = Add()([x_0, x])
return x
def create_model():
# Implementation of WideResNet (depth = 16, width = 10) based on keras_contrib
# https://github.com/keras-team/keras-contrib/blob/master/keras_contrib/applications/wide_resnet.py
inputs = Input(shape=(28, 28, 1))
x = cbr(inputs, 16)
x = conv_block(x, 160, True)
x = conv_block(x, 160)
x = MaxPooling2D((2, 2))(x)
x = conv_block(x, 320, True)
x = conv_block(x, 320)
x = MaxPooling2D((2, 2))(x)
x = conv_block(x, 640, True)
x = conv_block(x, 640)
x = GlobalAveragePooling2D()(x)
outputs = Dense(num_classes, activation='softmax')(x)
model = tf.keras.models.Model(inputs, outputs)
opt = tf.keras.optimizers.SGD(lr=args.base_lr)
# Wrap the optimizer in a Horovod distributed optimizer
opt = hvd.DistributedOptimizer(opt)
model.compile(loss=tf.keras.losses.categorical_crossentropy,
optimizer=opt,
metrics=['accuracy'])
return model
if hvd.rank() == 0:
verbose = 1
else:
verbose = 0
# Input image dimensions
img_rows, img_cols = 28, 28
num_classes = 10
# Load Fashion MNIST data.
(x_train, y_train), (x_test, y_test) = fashion_mnist.load_data()
if K.image_data_format() == 'channels_first':
x_train = x_train.reshape(x_train.shape[0], 1, img_rows, img_cols)
x_test = x_test.reshape(x_test.shape[0], 1, img_rows, img_cols)
input_shape = (1, img_rows, img_cols)
else:
x_train = x_train.reshape(x_train.shape[0], img_rows, img_cols, 1)
x_test = x_test.reshape(x_test.shape[0], img_rows, img_cols, 1)
input_shape = (img_rows, img_cols, 1)
# Convert class vectors to binary class matrices
y_train = tf.keras.utils.to_categorical(y_train, num_classes)
y_test = tf.keras.utils.to_categorical(y_test, num_classes)
# Training data iterator.
train_gen = image.ImageDataGenerator(featurewise_center=True, featurewise_std_normalization=True,
horizontal_flip=True, width_shift_range=0.2, height_shift_range=0.2)
train_gen.fit(x_train)
train_iter = train_gen.flow(x_train, y_train, batch_size=args.batch_size)
# Validation data iterator.
test_gen = image.ImageDataGenerator(featurewise_center=True, featurewise_std_normalization=True)
test_gen.mean = train_gen.mean
test_gen.std = train_gen.std
test_iter = test_gen.flow(x_test, y_test, batch_size=args.val_batch_size)
callbacks = []
callbacks.append(tf.keras.callbacks.LearningRateScheduler(lr_schedule))
# Broadcast initial variable states from the first worker to all others.
callbacks.append(hvd.callbacks.BroadcastGlobalVariablesCallback(0))
# Average the metrics among workers at the end of every epoch.
callbacks.append(hvd.callbacks.MetricAverageCallback())
class PrintThroughput(tf.keras.callbacks.Callback):
def __init__(self, total_images=0):
self.total_images = total_images
def on_epoch_begin(self, epoch, logs=None):
self.epoch_start_time = time()
def on_epoch_end(self, epoch, logs={}):
epoch_time = time() - self.epoch_start_time
images_per_sec = round(self.total_images / epoch_time, 2)
print('Images/sec: {}'.format(images_per_sec))
if verbose:
callbacks.append(PrintThroughput(total_images=len(y_train)))
class StopAtAccuracy(tf.keras.callbacks.Callback):
def __init__(self, target=0.85, patience=2, verbose=0):
self.target = target
self.patience = patience
self.verbose = verbose
self.stopped_epoch = 0
self.met_target = 0
def on_epoch_end(self, epoch, logs=None):
if logs.get('val_accuracy') > self.target:
self.met_target += 1
else:
self.met_target = 0
if self.met_target >= self.patience:
self.stopped_epoch = epoch
self.model.stop_training = True
def on_train_end(self, logs=None):
if self.stopped_epoch > 0 and self.verbose == 1:
print('Early stopping after epoch {}'.format(self.stopped_epoch + 1))
callbacks.append(StopAtAccuracy(target=args.target_accuracy, patience=args.patience, verbose=verbose))
class PrintTotalTime(tf.keras.callbacks.Callback):
def on_train_begin(self, logs=None):
self.start_time = time()
def on_epoch_end(self, epoch, logs=None):
total_time = round(time() - self.start_time, 2)
print("Cumulative training time after epoch {}: {}".format(epoch + 1, total_time))
def on_train_end(self, logs=None):
total_time = round(time() - self.start_time, 2)
print("Cumulative training time: {}".format(total_time))
if verbose:
callbacks.append(PrintTotalTime())
class SaveTrainingData(tf.keras.callbacks.Callback):
def __init__(self, data_filepath=''):
self.data_filepath = data_filepath
def on_train_begin(self, logs=None):
file = open(self.data_filepath, 'w', newline='')
writer = csv.writer(file)
writer.writerow(['time', 'val_accuracy'])
writer.writerow([0.0, 0.0])
file.close()
self.train_start_time = time()
def on_epoch_end(self, epoch, logs={}):
total_time = time() - self.train_start_time
file = open(self.data_filepath, 'a')
writer = csv.writer(file)
writer.writerow([round(total_time,1), round(logs['val_accuracy'], 4)])
file.close()
# Save the training data.
if hvd.rank() == 0:
data_filepath = "training_data/{}ranks-{}bs-{}lr.csv".format(hvd.size(), args.batch_size, args.base_lr)
callbacks.append(SaveTrainingData(data_filepath=data_filepath))
# Create the model.
model = create_model()
# Train the model.
model.fit(train_iter,
steps_per_epoch=len(train_iter) // hvd.size(),
callbacks=callbacks,
epochs=args.epochs,
verbose=verbose,
workers=4,
initial_epoch=0,
validation_data=test_iter,
validation_steps=3 * len(test_iter) // hvd.size())
# Evaluate the model on the full data set.
score = model.evaluate(test_iter, steps=len(test_iter), workers=4, verbose=verbose)
if verbose:
print('Test loss:', score[0])
print('Test accuracy:', score[1])
add_momenum.py：
import argparse
import tensorflow as tf
from tensorflow.keras import backend as K
from tensorflow.keras.preprocessing import image
from tensorflow.keras.datasets import fashion_mnist
from tensorflow.keras.layers import Input, Conv2D, BatchNormalization, Dense, \
Add, Activation, Dropout, MaxPooling2D, GlobalAveragePooling2D
import numpy as np
import os
from time import time
import horovod.tensorflow.keras as hvd
import csv
# Initialize Horovod
hvd.init()
# Pin to a GPU
gpus = tf.config.experimental.list_physical_devices('GPU')
if gpus:
tf.config.experimental.set_memory_growth(gpus[hvd.local_rank()], True)
tf.config.experimental.set_visible_devices(gpus[hvd.local_rank()], 'GPU')
# Parse input arguments
parser = argparse.ArgumentParser(description='Fashion MNIST Example',
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument('--log-dir', default='./logs',
help='tensorboard log directory')
parser.add_argument('--batch-size', type=int, default=32,
help='input batch size for training')
parser.add_argument('--val-batch-size', type=int, default=32,
help='input batch size for validation')
parser.add_argument('--epochs', type=int, default=40,
help='number of epochs to train')
parser.add_argument('--base-lr', type=float, default=0.01,
help='learning rate for a single GPU')
parser.add_argument('--momentum', type=float, default=0.9,
help='SGD momentum')
parser.add_argument('--wd', type=float, default=0.000005,
help='weight decay')
parser.add_argument('--target-accuracy', type=float, default=.85,
help='Target accuracy to stop training')
parser.add_argument('--patience', type=float, default=2,
help='Number of epochs that meet target before stopping')
args = parser.parse_args()
# Define a function for a simple learning rate decay over time
def lr_schedule(epoch):
if epoch < 15:
return args.base_lr
if epoch < 25:
return 1e-1 * args.base_lr
if epoch < 35:
return 1e-2 * args.base_lr
return 1e-3 * args.base_lr
# Define the function that creates the model
def cbr(x, conv_size):
channel_axis = 1 if K.image_data_format() == 'channels_first' else -1
x = Conv2D(conv_size, (3,3), padding='same')(x)
x = BatchNormalization(axis=channel_axis)(x)
x = Activation('relu')(x)
return x
def conv_block(x, conv_size, scale_input = False):
x_0 = x
if scale_input:
x_0 = Conv2D(conv_size, (1, 1), activation='linear', padding='same')(x_0)
x = cbr(x, conv_size)
x = Dropout(0.01)(x)
x = cbr(x, conv_size)
x = Add()([x_0, x])
return x
def create_model():
# Implementation of WideResNet (depth = 16, width = 10) based on keras_contrib
# https://github.com/keras-team/keras-contrib/blob/master/keras_contrib/applications/wide_resnet.py
inputs = Input(shape=(28, 28, 1))
x = cbr(inputs, 16)
x = conv_block(x, 160, True)
x = conv_block(x, 160)
x = MaxPooling2D((2, 2))(x)
x = conv_block(x, 320, True)
x = conv_block(x, 320)
x = MaxPooling2D((2, 2))(x)
x = conv_block(x, 640, True)
x = conv_block(x, 640)
x = GlobalAveragePooling2D()(x)
outputs = Dense(num_classes, activation='softmax')(x)
model = tf.keras.models.Model(inputs, outputs)
opt = tf.keras.optimizers.SGD(lr=args.base_lr, momentum=args.momentum)
# Wrap the optimizer in a Horovod distributed optimizer
opt = hvd.DistributedOptimizer(opt)
model.compile(loss=tf.keras.losses.categorical_crossentropy,
optimizer=opt,
metrics=['accuracy'])
return model
if hvd.rank() == 0:
verbose = 1
else:
verbose = 0
# Input image dimensions
img_rows, img_cols = 28, 28
num_classes = 10
# Load Fashion MNIST data.
(x_train, y_train), (x_test, y_test) = fashion_mnist.load_data()
if K.image_data_format() == 'channels_first':
x_train = x_train.reshape(x_train.shape[0], 1, img_rows, img_cols)
x_test = x_test.reshape(x_test.shape[0], 1, img_rows, img_cols)
input_shape = (1, img_rows, img_cols)
else:
x_train = x_train.reshape(x_train.shape[0], img_rows, img_cols, 1)
x_test = x_test.reshape(x_test.shape[0], img_rows, img_cols, 1)
input_shape = (img_rows, img_cols, 1)
# Convert class vectors to binary class matrices
y_train = tf.keras.utils.to_categorical(y_train, num_classes)
y_test = tf.keras.utils.to_categorical(y_test, num_classes)
# Training data iterator.
train_gen = image.ImageDataGenerator(featurewise_center=True, featurewise_std_normalization=True,
horizontal_flip=True, width_shift_range=0.2, height_shift_range=0.2)
train_gen.fit(x_train)
train_iter = train_gen.flow(x_train, y_train, batch_size=args.batch_size)
# Validation data iterator.
test_gen = image.ImageDataGenerator(featurewise_center=True, featurewise_std_normalization=True)
test_gen.mean = train_gen.mean
test_gen.std = train_gen.std
test_iter = test_gen.flow(x_test, y_test, batch_size=args.val_batch_size)
callbacks = []
callbacks.append(tf.keras.callbacks.LearningRateScheduler(lr_schedule))
# Broadcast initial variable states from the first worker to all others.
callbacks.append(hvd.callbacks.BroadcastGlobalVariablesCallback(0))
# Average the metrics among workers at the end of every epoch.
callbacks.append(hvd.callbacks.MetricAverageCallback())
class PrintThroughput(tf.keras.callbacks.Callback):
def __init__(self, total_images=0):
self.total_images = total_images
def on_epoch_begin(self, epoch, logs=None):
self.epoch_start_time = time()
def on_epoch_end(self, epoch, logs={}):
epoch_time = time() - self.epoch_start_time
images_per_sec = round(self.total_images / epoch_time, 2)
print('Images/sec: {}'.format(images_per_sec))
if verbose:
callbacks.append(PrintThroughput(total_images=len(y_train)))
class StopAtAccuracy(tf.keras.callbacks.Callback):
def __init__(self, target=0.85, patience=2, verbose=0):
self.target = target
self.patience = patience
self.verbose = verbose
self.stopped_epoch = 0
self.met_target = 0
def on_epoch_end(self, epoch, logs=None):
if logs.get('val_accuracy') > self.target:
self.met_target += 1
else:
self.met_target = 0
if self.met_target >= self.patience:
self.stopped_epoch = epoch
self.model.stop_training = True
def on_train_end(self, logs=None):
if self.stopped_epoch > 0 and self.verbose == 1:
print('Early stopping after epoch {}'.format(self.stopped_epoch + 1))
callbacks.append(StopAtAccuracy(target=args.target_accuracy, patience=args.patience, verbose=verbose))
class PrintTotalTime(tf.keras.callbacks.Callback):
def on_train_begin(self, logs=None):
self.start_time = time()
def on_epoch_end(self, epoch, logs=None):
total_time = round(time() - self.start_time, 2)
print("Cumulative training time after epoch {}: {}".format(epoch + 1, total_time))
def on_train_end(self, logs=None):
total_time = round(time() - self.start_time, 2)
print("Cumulative training time: {}".format(total_time))
if verbose:
callbacks.append(PrintTotalTime())
class SaveTrainingData(tf.keras.callbacks.Callback):
def __init__(self, data_filepath=''):
self.data_filepath = data_filepath
def on_train_begin(self, logs=None):
file = open(self.data_filepath, 'w', newline='')
writer = csv.writer(file)
writer.writerow(['time', 'val_accuracy'])
writer.writerow([0.0, 0.0])
file.close()
self.train_start_time = time()
def on_epoch_end(self, epoch, logs={}):
total_time = time() - self.train_start_time
file = open(self.data_filepath, 'a')
writer = csv.writer(file)
writer.writerow([round(total_time,1), round(logs['val_accuracy'], 4)])
file.close()
# Save the training data.
if hvd.rank() == 0:
data_filepath = "training_data/{}ranks-{}bs-{}lr-{}m.csv".format(hvd.size(), args.batch_size, args.base_lr, args.momentum)
callbacks.append(SaveTrainingData(data_filepath=data_filepath))
# Create the model.
model = create_model()
# Train the model.
model.fit(train_iter,
steps_per_epoch=len(train_iter) // hvd.size(),
callbacks=callbacks,
epochs=args.epochs,
verbose=verbose,
workers=4,
initial_epoch=0,
validation_data=test_iter,
validation_steps=3 * len(test_iter) // hvd.size())
# Evaluate the model on the full data set.
score = model.evaluate(test_iter, steps=len(test_iter), workers=4, verbose=verbose)
if verbose:
print('Test loss:', score[0])
print('Test accuracy:', score[1])
参考资料
- Nividia的课件《用多 GPU 训练神经网络》
药企，独角兽，苏州。团队长期招人，感兴趣的都可以发邮件聊聊：tiehan@sina.cn
![]()
个人公众号，比较懒，很少更新，可以在上面提问题，如果回复不及时，可发邮件给我： tiehan@sina.cn
