fault_tolerance.ipynb

Kernel: Python 3

Copyright 2021 The TensorFlow Authors.

#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Cómo migrar el mecanismo de tolerancia ante errores

Ver en TensorFlow.org

Ejecutar en Google Colab

Ver código fuente en GitHub

Descargar bloc de notas

La tolerancia ante errores se refiere a un mecanismo de guardado periódico de los estados de objetos rastreables, como parámetros y modelos. Esto permite recuperarlos en caso de que falle el programa o la máquina durante el entrenamiento.

En esta guía primero se muestra cómo incorporar la tolerancia ante errores al entrenamiento con tf.estimator.Estimator en TensorFlow 1 especificando el ahorro de métricas con tf.estimator.RunConfig. Después, aprenderá a implementar la tolerancia ante errores para el entrenamiento en TensorFlow 2 de dos maneras:

• Si utiliza la API de Keras Model.fit, puede transferirle la retrollamada con tf.keras.callbacks.BackupAndRestore.

• Si utiliza un bucle de entrenamiento personalizado (con tf.GradientTape), puede guardar arbitrariamente los puntos de verificación utilizando las API tf.train.Checkpoint y tf.train.CheckpointManager.

Ambos métodos harán una copia de seguridad y restaurarán los estados de entrenamiento en los archivos de puntos de verificación.

Preparación

Instale tf-nightly, ya que la frecuencia de guardado de puntos de verificación en un paso concreto con el argumento save_freq en tf.keras.callbacks.BackupAndRestore se introduce a partir de TensorFlow 2.10:

!pip install tf-nightly

import tensorflow.compat.v1 as tf1
import tensorflow as tf
import numpy as np
import tempfile
import time

mnist = tf.keras.datasets.mnist

(x_train, y_train),(x_test, y_test) = mnist.load_data()
x_train, x_test = x_train / 255.0, x_test / 255.0

TensorFlow 1: Guardar los puntos de verificación con tf.estimator.RunConfig

En TensorFlow 1, puede configurar un tf.estimator para guardar puntos de verificación en cada paso configurando tf.estimator.RunConfig.

En este ejemplo, comience escribiendo un hook que arroje artificialmente un error durante el quinto punto de verificación:

class InterruptHook(tf1.train.SessionRunHook):
  # A hook for artificially interrupting training.
  def begin(self):
    self._step = -1

  def before_run(self, run_context):
    self._step += 1

  def after_run(self, run_context, run_values):
    if self._step == 5:
      raise RuntimeError('Interruption')

A continuación, configure tf.estimator.Estimator para guardar cada punto de verificación y utilice el conjunto de datos MNIST:

feature_columns = [tf1.feature_column.numeric_column("x", shape=[28, 28])]
config = tf1.estimator.RunConfig(save_summary_steps=1,
                                 save_checkpoints_steps=1)

path = tempfile.mkdtemp()

classifier = tf1.estimator.DNNClassifier(
    feature_columns=feature_columns,
    hidden_units=[256, 32],
    optimizer=tf1.train.AdamOptimizer(0.001),
    n_classes=10,
    dropout=0.2,
    model_dir=path,
    config = config
)

train_input_fn = tf1.estimator.inputs.numpy_input_fn(
    x={"x": x_train},
    y=y_train.astype(np.int32),
    num_epochs=10,
    batch_size=50,
    shuffle=True,
)

Comience a entrenar el modelo. Se producirá una excepción artificial mediante el hook que definió anteriormente.

try:
  classifier.train(input_fn=train_input_fn,
                   hooks=[InterruptHook()],
                   max_steps=10)
except Exception as e:
  print(f'{type(e).__name__}:{e}')

Vuelva a construir el tf.estimator.Estimator utilizando el último punto de verificación guardado y continúe con el entrenamiento:

classifier = tf1.estimator.DNNClassifier(
    feature_columns=feature_columns,
    hidden_units=[256, 32],
    optimizer=tf1.train.AdamOptimizer(0.001),
    n_classes=10,
    dropout=0.2,
    model_dir=path,
    config = config
)
classifier.train(input_fn=train_input_fn,
                   max_steps = 10)

TensorFlow 2: Copia de seguridad y restauración con una retrollamada y Model.fit

En TensorFlow 2, si utiliza la API Keras Model.fit para el entrenamiento, puede proporcionar la retrollamada tf.keras.callbacks.BackupAndRestore para incorporar la funcionalidad de tolerancia ante errores.

Para ayudar a demostrar esto, en primer lugar comience por definir una clase de la Callback de Keras que produzca artificialmente un error durante el cuarto punto de verificación de la época:

class InterruptAtEpoch(tf.keras.callbacks.Callback):
  # A callback for artificially interrupting training.
  def __init__(self, interrupting_epoch=3):
    self.interrupting_epoch = interrupting_epoch

  def on_epoch_end(self, epoch, log=None):
    if epoch == self.interrupting_epoch:
      raise RuntimeError('Interruption')

A continuación, defina y cree una instancia de un modelo simple de Keras, defina la función de pérdida, llame a Model.compile, y configure una retrollamada tf.keras.callbacks.BackupAndRestore que guardará los puntos de verificación en un directorio temporal en los límites de las épocas:

def create_model():
  return tf.keras.models.Sequential([
    tf.keras.layers.Flatten(input_shape=(28, 28)),
    tf.keras.layers.Dense(512, activation='relu'),
    tf.keras.layers.Dropout(0.2),
    tf.keras.layers.Dense(10)
  ])
loss = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)
model = create_model()
model.compile(optimizer='adam',
              loss=loss,
              metrics=['accuracy'])
log_dir = tempfile.mkdtemp()
backup_restore_callback = tf.keras.callbacks.BackupAndRestore(
    backup_dir = log_dir)

Comience a entrenar el modelo con Model.fit. Durante el entrenamiento, los puntos de verificación se guardarán gracias a tf.keras.callbacks.BackupAndRestore creado anteriormente, mientras que la clase InterruptAtEpoch generará una excepción artificial para simular un error después de la cuarta época.

try:
  model.fit(x=x_train,
            y=y_train,
            epochs=10,
            steps_per_epoch=100,
            validation_data=(x_test, y_test),
            callbacks=[backup_restore_callback, InterruptAtEpoch()])
except Exception as e:
  print(f'{type(e).__name__}:{e}')

Luego, cree una instancia del modelo Keras, llame a Model.compile, y continúe entrenando el modelo con Model.fit desde un punto de verificación guardado previamente:

model = create_model()
model.compile(optimizer='adam',
              loss=loss,
              metrics=['accuracy'],
              steps_per_execution=10)
model.fit(x=x_train,
            y=y_train,
            epochs=10,
            steps_per_epoch=100,
            validation_data=(x_test, y_test),
            callbacks=[backup_restore_callback])

Defina otra clase de la Callback que produzca artificialmente un error durante el paso 140:

class InterruptAtStep(tf.keras.callbacks.Callback):
  # A callback for artificially interrupting training.
  def __init__(self, interrupting_step=140):
    self.total_step_count = 0
    self.interrupting_step = interrupting_step

  def on_batch_begin(self, batch, logs=None):
    self.total_step_count += 1

  def on_batch_end(self, batch, logs=None):
    if self.total_step_count == self.interrupting_step:
      print("\nInterrupting at step count", self.total_step_count)
      raise RuntimeError('Interruption')

Nota: En esta sección se utilizan funciones que solo están disponibles en tf-nightly hasta que se publique Tensorflow 2.10.

Para asegurarse de que los puntos de verificación se guardan cada 30 pasos, establezca save_freq en BackupAndRestore de la retrollamada a 30. El InterruptAtStep producirá una excepción artificial para simular un error en la época 1 y el paso 40 (número total de pasos 140). El punto de verificación se guardaría por última vez en la época 1 y el paso 20.

log_dir_2 = tempfile.mkdtemp()

backup_restore_callback = tf.keras.callbacks.BackupAndRestore(
    backup_dir = log_dir_2, save_freq=30
)
model = create_model()
model.compile(optimizer='adam',
              loss=loss,
              metrics=['accuracy'])
try:
  model.fit(x=x_train,
            y=y_train,
            epochs=10,
            steps_per_epoch=100,
            validation_data=(x_test, y_test),
            callbacks=[backup_restore_callback, InterruptAtStep()])
except Exception as e:
  print(f'{type(e).__name__}:{e}')

Posteriormente, cree una instancia del modelo de Keras, llame a Model.compile, y continúe entrenando el modelo con Model.fit desde un punto de verificación guardado previamente. Observe que el entrenamiento inicia desde la época 2 y el paso 21.

model = create_model()
model.compile(optimizer='adam',
              loss=loss,
              metrics=['accuracy'],
              steps_per_execution=10)
model.fit(x=x_train,
            y=y_train,
            epochs=10,
            steps_per_epoch=100,
            validation_data=(x_test, y_test),
            callbacks=[backup_restore_callback])

TensorFlow 2: Escribiendo puntos de verificación manuales con un bucle de entrenamiento personalizado

Si utiliza un bucle de entrenamiento personalizado en TensorFlow 2, puede implementar un mecanismo de tolerancia ante errores con las API tf.train.Checkpoint y tf.train.CheckpointManager.

En este ejemplo se muestra cómo hacerlo:

• Utilice un objeto tf.train.Checkpoint para crear manualmente un punto de verificación, donde los objetos rastreables que desea guardar se establezcan como atributos.

• Utilice un tf.train.CheckpointManager para administrar varios puntos de verificación.

Comience por definir y crear instancias del modelo Keras, el optimizador y la función de pérdida. Después, cree un Checkpoint que administre dos objetos con estados rastreables (el modelo y el optimizador), así como un CheckpointManager para registrar y mantener varios puntos de verificación en un directorio temporal.

model = create_model()
optimizer = tf.keras.optimizers.SGD(learning_rate=0.001)
loss_fn = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)
log_dir = tempfile.mkdtemp()
epochs = 5
steps_per_epoch = 5

checkpoint = tf.train.Checkpoint(model=model, optimizer=optimizer)
checkpoint_manager = tf.train.CheckpointManager(
            checkpoint, log_dir, max_to_keep=2)

Ahora, implemente un bucle de entrenamiento personalizado donde después de la primera época cada vez que comienza una nueva época se cargará el último punto de verificación:

for epoch in range(epochs):
  if epoch > 0:
      tf.train.load_checkpoint(save_path)
  print(f"\nStart of epoch {epoch}")

  for step in range(steps_per_epoch):
    with tf.GradientTape() as tape:

      logits = model(x_train, training=True)
      loss_value = loss_fn(y_train, logits)

      grads = tape.gradient(loss_value, model.trainable_weights)
      optimizer.apply_gradients(zip(grads, model.trainable_weights))

    save_path = checkpoint_manager.save()
    print(f"Checkpoint saved to {save_path}")
    print(f"Training loss at step {step}: {loss_value}")

Siguientes pasos

Para obtener más información sobre la tolerancia ante errores y la verificación de puntos en TensorFlow 2, consulte la siguiente documentación:

• La documentación de la API de retrollamadas tf.keras.callbacks.BackupAndRestore.

• La documentación de las API tf.train.Checkpoint y tf.train.CheckpointManager.

• La guía Puntos de verificación del entrenamiento, incluyendo la sección Puntos de verificación de la escritura.

También podrá encontrar útil el siguiente material relacionado con el entrenamiento distribuido:

• La sección Tolerancia ante errores del tutorial Entrenamiento multi-trabajador con Keras.

• La sección Error de la tarea en el tutorial Entrenamiento del servidor de parámetros.