अनुमानक के साथ बहु-कार्यकर्ता प्रशिक्षण

TensorFlow.org पर देखें

Google Colab में चलाएं

GitHub पर स्रोत देखें

नोटबुक डाउनलोड करें

अवलोकन

यह ट्यूटोरियल दर्शाता है कि tf.distribute.Strategy tf.estimator उपयोग कैसे किया जा सकता है। यदि आप tf.estimator का उपयोग करके अपना कोड लिखते हैं, और आप उच्च प्रदर्शन वाली एकल मशीन से आगे बढ़ने में रुचि रखते हैं, तो यह ट्यूटोरियल आपके लिए है।

आरंभ करने से पहले, कृपया वितरण रणनीति मार्गदर्शिका पढ़ें। मल्टी-जीपीयू प्रशिक्षण ट्यूटोरियल भी प्रासंगिक है, क्योंकि यह ट्यूटोरियल उसी मॉडल का उपयोग करता है।

सेट अप

सबसे पहले, TensorFlow और आवश्यक आयात सेटअप करें।

import tensorflow_datasets as tfds
import tensorflow as tf

import os, json

tf.compat.v1.disable_eager_execution()

इनपुट फ़ंक्शन

यह ट्यूटोरियल TensorFlow डेटासेट से MNIST डेटासेट का उपयोग करता है। यहां कोड एक महत्वपूर्ण अंतर के साथ बहु-जीपीयू प्रशिक्षण ट्यूटोरियल के समान है: बहु-कार्यकर्ता प्रशिक्षण के लिए अनुमानक का उपयोग करते समय, मॉडल अभिसरण सुनिश्चित करने के लिए डेटासेट को श्रमिकों की संख्या से विभाजित करना आवश्यक है। इनपुट डेटा को वर्कर इंडेक्स द्वारा शार्प किया जाता है, ताकि प्रत्येक कार्यकर्ता डेटासेट के 1/num_workers अलग-अलग हिस्सों को प्रोसेस करे।

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)

अभिसरण प्राप्त करने का एक अन्य उचित तरीका यह होगा कि प्रत्येक कार्यकर्ता पर अलग-अलग बीजों के साथ डेटासेट में फेरबदल किया जाए।

बहु-कार्यकर्ता विन्यास

इस ट्यूटोरियल में प्रमुख अंतरों में से एक ( मल्टी-जीपीयू ट्रेनिंग ट्यूटोरियल की तुलना में) मल्टी-वर्कर सेटअप है। TF_CONFIG पर्यावरण चर प्रत्येक कार्यकर्ता को क्लस्टर कॉन्फ़िगरेशन निर्दिष्ट करने का मानक तरीका है जो क्लस्टर का हिस्सा है।

TF_CONFIG के दो घटक हैं: cluster और task । cluster पूरे क्लस्टर के बारे में जानकारी प्रदान करता है, अर्थात् क्लस्टर में कार्यकर्ता और पैरामीटर सर्वर। task वर्तमान कार्य के बारे में जानकारी प्रदान करता है। पहला घटक cluster में सभी श्रमिकों और पैरामीटर सर्वरों के लिए समान है, और दूसरा घटक task प्रत्येक कार्यकर्ता और पैरामीटर सर्वर पर अलग है और इसका अपना type और index निर्दिष्ट करता है। इस उदाहरण में, कार्य type worker है और कार्य index 0 है।

उदाहरण के लिए, यह ट्यूटोरियल दिखाता है कि localhost पर 2 श्रमिकों के साथ TF_CONFIG कैसे सेट करें। व्यवहार में, आप बाहरी आईपी पते और पोर्ट पर कई कर्मचारी बनाएंगे, और प्रत्येक कार्यकर्ता पर उचित रूप से TF_CONFIG सेट करेंगे, अर्थात कार्य index को संशोधित करेंगे।

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

मॉडल को परिभाषित करें

प्रशिक्षण के लिए परतें, अनुकूलक और हानि फ़ंक्शन लिखें। यह ट्यूटोरियल मल्टी-जीपीयू ट्रेनिंग ट्यूटोरियल के समान केरस लेयर्स वाले मॉडल को परिभाषित करता है।

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()))

MultiWorkerप्रतिबिंबित रणनीति

मॉडल को प्रशिक्षित करने के लिए, tf.distribute.experimental.MultiWorkerMirroredStrategy के उदाहरण का उपयोग करें। MultiWorkerMirroredStrategy सभी कर्मचारियों के लिए प्रत्येक डिवाइस पर मॉडल की परतों में सभी चर की प्रतियां बनाता है। यह सामूहिक संचार के लिए एक TensorFlow op, CollectiveOps का उपयोग करता है, ग्रेडिएंट को एकत्रित करने और चर को सिंक में रखने के लिए। इस रणनीति के बारे में tf. tf.distribute.Strategy मार्गदर्शिका में अधिक विवरण हैं।

strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()

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

मॉडल को प्रशिक्षित और मूल्यांकन करें

इसके बाद, अनुमानक के लिए RunConfig में वितरण रणनीति निर्दिष्ट करें, और tf.estimator.train_and_evaluate को लागू करके प्रशिक्षण और मूल्यांकन करें। यह ट्यूटोरियल train_distribute के माध्यम से रणनीति निर्दिष्ट करके केवल प्रशिक्षण वितरित करता है। मूल्यांकन को eval_distribute के माध्यम से वितरित करना भी संभव है।

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)
)

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}, [])

प्रशिक्षण प्रदर्शन का अनुकूलन करें

अब आपके पास tf.distribute.Strategy द्वारा संचालित एक मॉडल और एक बहु-कार्यकर्ता सक्षम अनुमानक है। बहु-कार्यकर्ता प्रशिक्षण के प्रदर्शन को अनुकूलित करने के लिए आप निम्नलिखित तकनीकों को आजमा सकते हैं:

• बैच आकार बढ़ाएँ: यहाँ निर्दिष्ट बैच आकार प्रति-GPU है। सामान्य तौर पर, GPU मेमोरी में फिट होने वाले सबसे बड़े बैच आकार की सलाह दी जाती है।
• कास्ट वेरिएबल्स: यदि संभव हो तो वेरिएबल को tf.float पर कास्ट करें। आधिकारिक ResNet मॉडल में एक उदाहरण शामिल है कि यह कैसे किया जा सकता है।

• सामूहिक संचार का उपयोग करें: MultiWorkerMirroredStrategy कई सामूहिक संचार कार्यान्वयन प्रदान करता है।

  • RING क्रॉस-होस्ट संचार परत के रूप में जीआरपीसी का उपयोग करते हुए रिंग-आधारित सामूहिकता को लागू करता है।
  • NCCL सामूहिकों को लागू करने के लिए एनवीडिया के एनसीसीएल का उपयोग करता है।
  • AUTO रनटाइम के लिए पसंद को टाल देता है।

  सामूहिक कार्यान्वयन का सबसे अच्छा विकल्प GPU की संख्या और प्रकार और क्लस्टर में नेटवर्क इंटरकनेक्ट पर निर्भर करता है। स्वचालित पसंद को ओवरराइड करने के लिए, MultiWorkerMirroredStrategy के कंस्ट्रक्टर के communication पैरामीटर के लिए एक मान्य मान निर्दिष्ट करें, उदाहरण के लिए communication=tf.distribute.experimental.CollectiveCommunication.NCCL ।

अन्य रणनीतियों और उपकरणों के बारे में अधिक जानने के लिए गाइड में प्रदर्शन अनुभाग पर जाएँ जिनका उपयोग आप अपने TensorFlow मॉडल के प्रदर्शन को अनुकूलित करने के लिए कर सकते हैं।

अन्य कोड उदाहरण

1. कुबेरनेट्स टेम्प्लेट का उपयोग करके टेंसरफ़्लो/पारिस्थितिकी तंत्र में बहु-कार्यकर्ता प्रशिक्षण के लिए एंड-टू-एंड उदाहरण । यह उदाहरण केरस मॉडल से शुरू होता है और इसे tf.keras.estimator.model_to_estimator API का उपयोग करके एक अनुमानक में परिवर्तित करता है।
2. आधिकारिक मॉडल , जिनमें से कई को कई वितरण रणनीतियों को चलाने के लिए कॉन्फ़िगर किया जा सकता है।