使用TensorFlow进行多GPU分布式训练

import os

os.environ["KERAS_BACKEND"] = "tensorflow"

import tensorflow as tf
import keras

# 创建一个MirroredStrategy。
strategy = tf.distribute.MirroredStrategy()
print('Number of devices: {}'.format(strategy.num_replicas_in_sync))

# 打开一个策略作用域。
with strategy.scope():
    # 所有创建变量的操作都应该在策略作用域内。
    # 通常这只是模型构造和`compile()`。
    model = Model(...)
    model.compile(...)

    # 在所有可用设备上训练模型。
    model.fit(train_dataset, validation_data=val_dataset, ...)

    # 在所有可用设备上测试模型。
    model.evaluate(test_dataset)

def get_compiled_model():
    # 创建一个简单的2层全连接神经网络。
    inputs = keras.Input(shape=(784,))
    x = keras.layers.Dense(256, activation="relu")(inputs)
    x = keras.layers.Dense(256, activation="relu")(x)
    outputs = keras.layers.Dense(10)(x)
    model = keras.Model(inputs, outputs)
    model.compile(
        optimizer=keras.optimizers.Adam(),
        loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),
        metrics=[keras.metrics.SparseCategoricalAccuracy()],
    )
    return model


def get_dataset():
    batch_size = 32
    num_val_samples = 10000

    # 返回以[`tf.data.Dataset`](https://www.tensorflow.org/api_docs/python/tf/data/Dataset)形式表示的MNIST数据集。
    (x_train, y_train), (x_test, y_test) = keras.datasets.mnist.load_data()

    # 预处理数据（这些是Numpy数组）
    x_train = x_train.reshape(-1, 784).astype("float32") / 255
    x_test = x_test.reshape(-1, 784).astype("float32") / 255
    y_train = y_train.astype("float32")
    y_test = y_test.astype("float32")

    # 保留num_val_samples个样本用于验证
    x_val = x_train[-num_val_samples:]
    y_val = y_train[-num_val_samples:]
    x_train = x_train[:-num_val_samples]
    y_train = y_train[:-num_val_samples]
    return (
        tf.data.Dataset.from_tensor_slices((x_train, y_train)).batch(batch_size),
        tf.data.Dataset.from_tensor_slices((x_val, y_val)).batch(batch_size),
        tf.data.Dataset.from_tensor_slices((x_test, y_test)).batch(batch_size),
    )


# 创建一个MirroredStrategy。
strategy = tf.distribute.MirroredStrategy()
print("Number of devices: {}".format(strategy.num_replicas_in_sync))

# 打开一个策略范围。
with strategy.scope():
    # 所有创建变量的操作都应该在策略范围内。
    # 通常这只是模型构建和`compile()`。
    model = get_compiled_model()

    # 在所有可用设备上训练模型。
    train_dataset, val_dataset, test_dataset = get_dataset()
    model.fit(train_dataset, epochs=2, validation_data=val_dataset)

    # 在所有可用设备上测试模型。
    model.evaluate(test_dataset)

# 准备一个目录来存储所有的检查点。
checkpoint_dir = "./ckpt"
if not os.path.exists(checkpoint_dir):
    os.makedirs(checkpoint_dir)


def make_or_restore_model():
    # 要么恢复最新的模型，要么如果没有检查点可用则创建一个新的模型。
    checkpoints = [checkpoint_dir + "/" + name for name in os.listdir(checkpoint_dir)]
    if checkpoints:
        latest_checkpoint = max(checkpoints, key=os.path.getctime)
        print("从", latest_checkpoint, "恢复")
        return keras.models.load_model(latest_checkpoint)
    print("创建一个新的模型")
    return get_compiled_model()


def run_training(epochs=1):
    # 创建一个 MirroredStrategy。
    strategy = tf.distribute.MirroredStrategy()

    # 打开一个策略作用域并创建/恢复模型
    with strategy.scope():
        model = make_or_restore_model()

        callbacks = [
            # 这个回调每周期保存一个 SavedModel
            # 我们在文件夹名称中包含当前周期。
            keras.callbacks.ModelCheckpoint(
                filepath=checkpoint_dir + "/ckpt-{epoch}.keras",
                save_freq="epoch",
            )
        ]
        model.fit(
            train_dataset,
            epochs=epochs,
            callbacks=callbacks,
            validation_data=val_dataset,
            verbose=2,
        )


# 第一次运行时创建模型
run_training(epochs=1)

# 再次调用相同函数将从上次停止的地方继续
run_training(epochs=1)