Tahminci ile çok çalışan eğitimi

TensorFlow.org'da görüntüleyin

Google Colab'da çalıştırın

Kaynağı GitHub'da görüntüleyin

Not defterini indir

Uyarı: Tahminciler yeni kod için önerilmez. Tahminciler, doğru yazılması daha zor olan ve özellikle TF 2 koduyla birleştirildiğinde beklenmedik şekilde davranabilen v1.Session tarzı kodu çalıştırır. Tahminciler uyumluluk garantileri kapsamındadır, ancak güvenlik açıkları dışında herhangi bir düzeltme almayacaklardır. Ayrıntılar için geçiş kılavuzuna bakın.

genel bakış

Bu öğretici, tf.distribute.Strategy ile dağıtılmış çok çalışanlı eğitim için nasıl kullanılabileceğini tf.estimator . Kodunuzu tf.estimator kullanarak yazıyorsanız ve yüksek performanslı tek bir makinenin ötesine geçmek istiyorsanız, bu eğitim tam size göre.

Başlamadan önce lütfen dağıtım stratejisi kılavuzunu okuyun. Çoklu GPU eğitim öğreticisi de önemlidir, çünkü bu öğretici aynı modeli kullanır.

Kurmak

İlk olarak, TensorFlow'u ve gerekli içe aktarmaları kurun.

import tensorflow_datasets as tfds
import tensorflow as tf

import os, json

tf.compat.v1.disable_eager_execution()

Giriş işlevi

Bu öğretici, TensorFlow Veri Kümelerinden MNIST veri kümesini kullanır. Buradaki kod, tek bir önemli farkla çoklu GPU eğitim öğreticisine benzer: Çok çalışanlı eğitim için Tahminci kullanılırken, model yakınsamasını sağlamak için veri kümesini çalışan sayısına göre parçalamak gerekir. Girdi verileri, çalışan dizini tarafından bölünür, böylece her çalışan, veri kümesinin 1/num_workers farklı bölümlerini işler.

BUFFER_SIZE = 10000
BATCH_SIZE = 64

def input_fn(mode, input_context=None):
  datasets, info = tfds.load(name='mnist',
                                with_info=True,
                                as_supervised=True)
  mnist_dataset = (datasets['train'] if mode == tf.estimator.ModeKeys.TRAIN else
                   datasets['test'])

  def scale(image, label):
    image = tf.cast(image, tf.float32)
    image /= 255
    return image, label

  if input_context:
    mnist_dataset = mnist_dataset.shard(input_context.num_input_pipelines,
                                        input_context.input_pipeline_id)
  return mnist_dataset.map(scale).cache().shuffle(BUFFER_SIZE).batch(BATCH_SIZE)

Yakınsamaya ulaşmak için başka bir makul yaklaşım, veri setini her çalışanda farklı tohumlarla karıştırmak olacaktır.

Çok çalışanlı yapılandırma

Bu öğreticideki temel farklılıklardan biri ( çoklu GPU eğitim öğreticisiyle karşılaştırıldığında), çok çalışanlı kurulumdur. TF_CONFIG ortam değişkeni, kümenin parçası olan her çalışana küme yapılandırmasını belirtmenin standart yoludur.

TF_CONFIG iki bileşeni vardır: cluster ve task . cluster , kümenin tamamı, yani kümedeki işçiler ve parametre sunucuları hakkında bilgi sağlar. task , geçerli görev hakkında bilgi sağlar. İlk bileşen cluster , kümedeki tüm çalışanlar ve parametre sunucuları için aynıdır ve ikinci bileşen task , her çalışan ve parametre sunucusunda farklıdır ve kendi type ve index belirtir. Bu örnekte, görev type worker ve görev index 0 .

Örnekleme amacıyla, bu öğretici, localhost üzerinde 2 işçi ile bir TF_CONFIG nasıl ayarlanacağını gösterir. Pratikte, harici bir IP adresi ve bağlantı noktasında birden çok işçi oluşturacak ve her işçiye uygun şekilde TF_CONFIG ayarlayacaksınız, yani görev index değiştireceksiniz.

os.environ['TF_CONFIG'] = json.dumps({
    'cluster': {
        'worker': ["localhost:12345", "localhost:23456"]
    },
    'task': {'type': 'worker', 'index': 0}
})

Modeli tanımlayın

Eğitim için katmanları, optimize ediciyi ve kayıp fonksiyonunu yazın. Bu öğretici, modeli çoklu GPU eğitim öğreticisine benzer şekilde Keras katmanlarıyla tanımlar.

LEARNING_RATE = 1e-4
def model_fn(features, labels, mode):
  model = tf.keras.Sequential([
      tf.keras.layers.Conv2D(32, 3, activation='relu', input_shape=(28, 28, 1)),
      tf.keras.layers.MaxPooling2D(),
      tf.keras.layers.Flatten(),
      tf.keras.layers.Dense(64, activation='relu'),
      tf.keras.layers.Dense(10)
  ])
  logits = model(features, training=False)

  if mode == tf.estimator.ModeKeys.PREDICT:
    predictions = {'logits': logits}
    return tf.estimator.EstimatorSpec(labels=labels, predictions=predictions)

  optimizer = tf.compat.v1.train.GradientDescentOptimizer(
      learning_rate=LEARNING_RATE)
  loss = tf.keras.losses.SparseCategoricalCrossentropy(
      from_logits=True, reduction=tf.keras.losses.Reduction.NONE)(labels, logits)
  loss = tf.reduce_sum(loss) * (1. / BATCH_SIZE)
  if mode == tf.estimator.ModeKeys.EVAL:
    return tf.estimator.EstimatorSpec(mode, loss=loss)

  return tf.estimator.EstimatorSpec(
      mode=mode,
      loss=loss,
      train_op=optimizer.minimize(
          loss, tf.compat.v1.train.get_or_create_global_step()))

Çoklu ÇalışanYansıtmalıStrateji

Modeli eğitmek için tf.distribute.experimental.MultiWorkerMirroredStrategy örneğini kullanın. MultiWorkerMirroredStrategy , tüm çalışanlarda her cihazda modelin katmanlarındaki tüm değişkenlerin kopyalarını oluşturur. Degradeleri toplamak ve değişkenleri senkronize tutmak için toplu iletişim için bir TensorFlow op olan CollectiveOps kullanır. tf.distribute.Strategy kılavuzunda bu strateji hakkında daha fazla ayrıntı bulunmaktadır.

strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()

tutucu6 l10n-yer

WARNING:tensorflow:From /tmp/ipykernel_7505/349189047.py:1: _CollectiveAllReduceStrategyExperimental.__init__ (from tensorflow.python.distribute.collective_all_reduce_strategy) is deprecated and will be removed in a future version.
Instructions for updating:
use distribute.MultiWorkerMirroredStrategy instead
INFO:tensorflow:Single-worker MultiWorkerMirroredStrategy with local_devices = ('/device:GPU:0',), communication = CommunicationImplementation.AUTO

Modeli eğitin ve değerlendirin

Ardından, tahmin edici için RunConfig içinde dağıtım stratejisini belirtin ve tf.estimator.train_and_evaluate çağırarak eğitin ve değerlendirin. Bu öğretici, train_distribute aracılığıyla stratejiyi belirterek yalnızca eğitimi dağıtır. Değerlendirmeyi eval_distribute aracılığıyla dağıtmak da mümkündür.

config = tf.estimator.RunConfig(train_distribute=strategy)

classifier = tf.estimator.Estimator(
    model_fn=model_fn, model_dir='/tmp/multiworker', config=config)
tf.estimator.train_and_evaluate(
    classifier,
    train_spec=tf.estimator.TrainSpec(input_fn=input_fn),
    eval_spec=tf.estimator.EvalSpec(input_fn=input_fn)
)

tutucu8 l10n-yer

INFO:tensorflow:Initializing RunConfig with distribution strategies.
INFO:tensorflow:Not using Distribute Coordinator.
INFO:tensorflow:Using config: {'_model_dir': '/tmp/multiworker', '_tf_random_seed': None, '_save_summary_steps': 100, '_save_checkpoints_steps': None, '_save_checkpoints_secs': 600, '_session_config': allow_soft_placement: true
graph_options {
  rewrite_options {
    meta_optimizer_iterations: ONE
  }
}
, '_keep_checkpoint_max': 5, '_keep_checkpoint_every_n_hours': 10000, '_log_step_count_steps': 100, '_train_distribute': <tensorflow.python.distribute.collective_all_reduce_strategy._CollectiveAllReduceStrategyExperimental object at 0x7f3404234490>, '_device_fn': None, '_protocol': None, '_eval_distribute': None, '_experimental_distribute': None, '_experimental_max_worker_delay_secs': None, '_session_creation_timeout_secs': 7200, '_checkpoint_save_graph_def': True, '_service': None, '_cluster_spec': ClusterSpec({}), '_task_type': 'worker', '_task_id': 0, '_global_id_in_cluster': 0, '_master': '', '_evaluation_master': '', '_is_chief': True, '_num_ps_replicas': 0, '_num_worker_replicas': 1, '_distribute_coordinator_mode': None}
INFO:tensorflow:Not using Distribute Coordinator.
INFO:tensorflow:Running training and evaluation locally (non-distributed).
INFO:tensorflow:Start train and evaluate loop. The evaluate will happen after every checkpoint. Checkpoint frequency is determined based on RunConfig arguments: save_checkpoints_steps None or save_checkpoints_secs 600.
WARNING:tensorflow:From /tmpfs/src/tf_docs_env/lib/python3.7/site-packages/tensorflow_estimator/python/estimator/estimator.py:1244: StrategyBase.configure (from tensorflow.python.distribute.distribute_lib) is deprecated and will be removed in a future version.
Instructions for updating:
use `update_config_proto` instead.
INFO:tensorflow:The `input_fn` accepts an `input_context` which will be given by DistributionStrategy
INFO:tensorflow:Calling model_fn.
/tmpfs/src/tf_docs_env/lib/python3.7/site-packages/tensorflow/python/data/ops/dataset_ops.py:449: UserWarning: To make it possible to preserve tf.data options across serialization boundaries, their implementation has moved to be part of the TensorFlow graph. As a consequence, the options value is in general no longer known at graph construction time. Invoking this method in graph mode retains the legacy behavior of the original implementation, but note that the returned value might not reflect the actual value of the options.
  warnings.warn("To make it possible to preserve tf.data options across "
INFO:tensorflow:Calling model_fn.
INFO:tensorflow:Done calling model_fn.
INFO:tensorflow:Done calling model_fn.
INFO:tensorflow:Create CheckpointSaverHook.
INFO:tensorflow:Create CheckpointSaverHook.
WARNING:tensorflow:From /tmpfs/src/tf_docs_env/lib/python3.7/site-packages/tensorflow_estimator/python/estimator/util.py:95: DistributedIteratorV1.initialize (from tensorflow.python.distribute.v1.input_lib) is deprecated and will be removed in a future version.
Instructions for updating:
Use the iterator's `initializer` property instead.
WARNING:tensorflow:From /tmpfs/src/tf_docs_env/lib/python3.7/site-packages/tensorflow_estimator/python/estimator/util.py:95: DistributedIteratorV1.initialize (from tensorflow.python.distribute.v1.input_lib) is deprecated and will be removed in a future version.
Instructions for updating:
Use the iterator's `initializer` property instead.
INFO:tensorflow:Graph was finalized.
INFO:tensorflow:Graph was finalized.
INFO:tensorflow:Running local_init_op.
INFO:tensorflow:Running local_init_op.
INFO:tensorflow:Done running local_init_op.
INFO:tensorflow:Done running local_init_op.
INFO:tensorflow:Calling checkpoint listeners before saving checkpoint 0...
INFO:tensorflow:Calling checkpoint listeners before saving checkpoint 0...
INFO:tensorflow:Saving checkpoints for 0 into /tmp/multiworker/model.ckpt.
INFO:tensorflow:Saving checkpoints for 0 into /tmp/multiworker/model.ckpt.
INFO:tensorflow:Calling checkpoint listeners after saving checkpoint 0...
INFO:tensorflow:Calling checkpoint listeners after saving checkpoint 0...
2022-01-26 05:29:43.503603: W tensorflow/core/grappler/utils/graph_view.cc:836] No registered 'MultiDeviceIteratorFromStringHandle' OpKernel for GPU devices compatible with node { {node MultiDeviceIteratorFromStringHandle} }
    .  Registered:  device='CPU'

2022-01-26 05:29:43.504873: W tensorflow/core/grappler/utils/graph_view.cc:836] No registered 'MultiDeviceIteratorGetNextFromShard' OpKernel for GPU devices compatible with node { {node MultiDeviceIteratorGetNextFromShard} }
    .  Registered:  device='CPU'
INFO:tensorflow:loss = 2.292878, step = 0
INFO:tensorflow:loss = 2.292878, step = 0
INFO:tensorflow:global_step/sec: 173.275
INFO:tensorflow:global_step/sec: 173.275
INFO:tensorflow:loss = 2.29561, step = 100 (0.579 sec)
INFO:tensorflow:loss = 2.29561, step = 100 (0.579 sec)
INFO:tensorflow:global_step/sec: 189.057
INFO:tensorflow:global_step/sec: 189.057
INFO:tensorflow:loss = 2.2644367, step = 200 (0.529 sec)
INFO:tensorflow:loss = 2.2644367, step = 200 (0.529 sec)
INFO:tensorflow:global_step/sec: 193.075
INFO:tensorflow:global_step/sec: 193.075
INFO:tensorflow:loss = 2.2662685, step = 300 (0.517 sec)
INFO:tensorflow:loss = 2.2662685, step = 300 (0.517 sec)
INFO:tensorflow:global_step/sec: 199.957
INFO:tensorflow:global_step/sec: 199.957
INFO:tensorflow:loss = 2.2667098, step = 400 (0.500 sec)
INFO:tensorflow:loss = 2.2667098, step = 400 (0.500 sec)
INFO:tensorflow:global_step/sec: 204.217
INFO:tensorflow:global_step/sec: 204.217
INFO:tensorflow:loss = 2.251912, step = 500 (0.490 sec)
INFO:tensorflow:loss = 2.251912, step = 500 (0.490 sec)
INFO:tensorflow:global_step/sec: 201.747
INFO:tensorflow:global_step/sec: 201.747
INFO:tensorflow:loss = 2.2633677, step = 600 (0.496 sec)
INFO:tensorflow:loss = 2.2633677, step = 600 (0.496 sec)
INFO:tensorflow:global_step/sec: 206.079
INFO:tensorflow:global_step/sec: 206.079
INFO:tensorflow:loss = 2.2531767, step = 700 (0.485 sec)
INFO:tensorflow:loss = 2.2531767, step = 700 (0.485 sec)
INFO:tensorflow:global_step/sec: 231.299
INFO:tensorflow:global_step/sec: 231.299
INFO:tensorflow:loss = 2.2578738, step = 800 (0.433 sec)
INFO:tensorflow:loss = 2.2578738, step = 800 (0.433 sec)
INFO:tensorflow:global_step/sec: 657.044
INFO:tensorflow:global_step/sec: 657.044
INFO:tensorflow:loss = 2.2344787, step = 900 (0.150 sec)
INFO:tensorflow:loss = 2.2344787, step = 900 (0.150 sec)
INFO:tensorflow:Calling checkpoint listeners before saving checkpoint 938...
INFO:tensorflow:Calling checkpoint listeners before saving checkpoint 938...
INFO:tensorflow:Saving checkpoints for 938 into /tmp/multiworker/model.ckpt.
INFO:tensorflow:Saving checkpoints for 938 into /tmp/multiworker/model.ckpt.
INFO:tensorflow:Calling checkpoint listeners after saving checkpoint 938...
INFO:tensorflow:Calling checkpoint listeners after saving checkpoint 938...
INFO:tensorflow:Calling model_fn.
INFO:tensorflow:Calling model_fn.
INFO:tensorflow:Done calling model_fn.
INFO:tensorflow:Done calling model_fn.
INFO:tensorflow:Starting evaluation at 2022-01-26T05:29:56
INFO:tensorflow:Starting evaluation at 2022-01-26T05:29:56
INFO:tensorflow:Graph was finalized.
INFO:tensorflow:Graph was finalized.
INFO:tensorflow:Restoring parameters from /tmp/multiworker/model.ckpt-938
INFO:tensorflow:Restoring parameters from /tmp/multiworker/model.ckpt-938
INFO:tensorflow:Running local_init_op.
INFO:tensorflow:Running local_init_op.
INFO:tensorflow:Done running local_init_op.
INFO:tensorflow:Done running local_init_op.
INFO:tensorflow:Evaluation [10/100]
INFO:tensorflow:Evaluation [10/100]
INFO:tensorflow:Evaluation [20/100]
INFO:tensorflow:Evaluation [20/100]
INFO:tensorflow:Evaluation [30/100]
INFO:tensorflow:Evaluation [30/100]
INFO:tensorflow:Evaluation [40/100]
INFO:tensorflow:Evaluation [40/100]
INFO:tensorflow:Evaluation [50/100]
INFO:tensorflow:Evaluation [50/100]
INFO:tensorflow:Evaluation [60/100]
INFO:tensorflow:Evaluation [60/100]
INFO:tensorflow:Evaluation [70/100]
INFO:tensorflow:Evaluation [70/100]
INFO:tensorflow:Evaluation [80/100]
INFO:tensorflow:Evaluation [80/100]
INFO:tensorflow:Evaluation [90/100]
INFO:tensorflow:Evaluation [90/100]
INFO:tensorflow:Evaluation [100/100]
INFO:tensorflow:Evaluation [100/100]
INFO:tensorflow:Inference Time : 2.04637s
INFO:tensorflow:Inference Time : 2.04637s
INFO:tensorflow:Finished evaluation at 2022-01-26-05:29:58
INFO:tensorflow:Finished evaluation at 2022-01-26-05:29:58
INFO:tensorflow:Saving dict for global step 938: global_step = 938, loss = 2.234131
INFO:tensorflow:Saving dict for global step 938: global_step = 938, loss = 2.234131
INFO:tensorflow:Saving 'checkpoint_path' summary for global step 938: /tmp/multiworker/model.ckpt-938
INFO:tensorflow:Saving 'checkpoint_path' summary for global step 938: /tmp/multiworker/model.ckpt-938
INFO:tensorflow:Loss for final step: 1.10881.
INFO:tensorflow:Loss for final step: 1.10881.
({'loss': 2.234131, 'global_step': 938}, [])

Eğitim performansını optimize edin

Artık bir modeliniz ve tf.distribute.Strategy tarafından desteklenen çok çalışanlı bir Tahminciniz var. Çok işçili eğitimin performansını optimize etmek için aşağıdaki teknikleri deneyebilirsiniz:

Parti boyutunu artırın: Burada belirtilen parti boyutu GPU başınadır. Genel olarak, GPU belleğine uyan en büyük parti boyutu tavsiye edilir.
Değişkenleri yayınla : Mümkünse değişkenleri tf.float . Resmi ResNet modeli, bunun nasıl yapılabileceğinin bir örneğini içerir.
Toplu iletişim kullanın: MultiWorkerMirroredStrategy , çoklu toplu iletişim uygulamaları sağlar.
- RING , ana bilgisayarlar arası iletişim katmanı olarak gRPC kullanan halka tabanlı kolektifleri uygular.
- NCCL , kolektifleri uygulamak için Nvidia'nın NCCL'sini kullanır.
- AUTO , seçimi çalışma zamanına erteler.
Toplu uygulamanın en iyi seçimi, GPU'ların sayısına ve türüne ve kümedeki ağ ara bağlantısına bağlıdır. Otomatik seçimi geçersiz kılmak için, MultiWorkerMirroredStrategy communication parametresine geçerli bir değer belirtin, ör.communica communication=tf.distribute.experimental.CollectiveCommunication.NCCL .

TensorFlow modellerinizin performansını optimize etmek için kullanabileceğiniz diğer stratejiler ve araçlar hakkında daha fazla bilgi edinmek için kılavuzdaki Performans bölümünü ziyaret edin.

Diğer kod örnekleri

Kubernetes şablonlarını kullanarak tensorflow/ekosistemde çok çalışanlı eğitim için uçtan uca örnek . Bu örnek bir Keras modeliyle başlar ve onu tf.keras.estimator.model_to_estimator API'sini kullanarak bir Tahmin Ediciye dönüştürür.
Birçoğu birden fazla dağıtım stratejisini çalıştıracak şekilde yapılandırılabilen resmi modeller .