 ดูบน TensorFlow.org |  ทำงานใน Google Colab |  ดูแหล่งที่มาบน GitHub |  ดาวน์โหลดโน๊ตบุ๊ค |
ภาพรวม
บทช่วยสอนนี้สาธิตวิธีใช้ tf.distribute.Strategy สำหรับการฝึกอบรมผู้ปฏิบัติงานหลายคนแบบกระจายด้วย tf.estimator หากคุณเขียนโค้ดโดยใช้ tf.estimator และคุณสนใจที่จะขยายขนาดเกินกว่าเครื่องเดียวที่มีประสิทธิภาพสูง บทแนะนำนี้เหมาะสำหรับคุณ
ก่อนเริ่มต้น โปรดอ่านคู่มือ กลยุทธ์การจัดจำหน่าย บทช่วย สอนการฝึกอบรม multi-GPU ก็มีความเกี่ยวข้องเช่นกัน เนื่องจากบทช่วยสอนนี้ใช้โมเดลเดียวกัน
ติดตั้ง
ขั้นแรก ตั้งค่า TensorFlow และการนำเข้าที่จำเป็น
import tensorflow_datasets as tfds
import tensorflow as tf
import os, json
tf.compat.v1.disable_eager_execution()
ฟังก์ชันอินพุต
บทช่วยสอนนี้ใช้ชุดข้อมูล MNIST จาก TensorFlow Datasets โค้ดในที่นี้คล้ายกับ บทช่วยสอนการฝึกอบรมแบบหลาย GPU โดยมีข้อแตกต่างที่สำคัญเพียงข้อเดียว: เมื่อใช้เครื่องมือประมาณการสำหรับการฝึกอบรมผู้ปฏิบัติงานหลายคน จำเป็นต้องแบ่งกลุ่มข้อมูลชุดข้อมูลตามจำนวนผู้ปฏิบัติงานเพื่อให้แน่ใจว่ามีการบรรจบกันของแบบจำลอง ข้อมูลที่ป้อนเข้าจะถูกแบ่งโดยดัชนีผู้ปฏิบัติงาน เพื่อให้ผู้ปฏิบัติงานแต่ละคนประมวลผล 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)
แนวทางที่สมเหตุสมผลอีกวิธีหนึ่งในการบรรลุการบรรจบกันคือการสับเปลี่ยนชุดข้อมูลด้วยเมล็ดพันธุ์ที่แตกต่างกันในแต่ละคนงาน
การกำหนดค่าผู้ปฏิบัติงานหลายคน
ความแตกต่างที่สำคัญอย่างหนึ่งในบทช่วยสอนนี้ (เมื่อเทียบกับบทช่วย สอนการฝึกอบรม GPU หลายตัว ) คือการตั้งค่าผู้ปฏิบัติงานหลายคน ตัวแปรสภาพแวดล้อม TF_CONFIG เป็นวิธีมาตรฐานในการระบุการกำหนดค่าคลัสเตอร์ให้กับผู้ปฏิบัติงานแต่ละคนที่เป็นส่วนหนึ่งของคลัสเตอร์
มีสององค์ประกอบของ TF_CONFIG : cluster และ task cluster ให้ข้อมูลเกี่ยวกับคลัสเตอร์ทั้งหมด กล่าวคือผู้ปฏิบัติงานและเซิร์ฟเวอร์พารามิเตอร์ในคลัสเตอร์ task ให้ข้อมูลเกี่ยวกับงานปัจจุบัน cluster คอมโพเนนต์แรกจะเหมือนกันสำหรับผู้ปฏิบัติงานและเซิร์ฟเวอร์พารามิเตอร์ทั้งหมดในคลัสเตอร์ และ task คอมโพเนนต์ที่สองจะแตกต่างกันในแต่ละผู้ปฏิบัติงานและเซิร์ฟเวอร์พารามิเตอร์ และระบุ type และ index ของตนเอง ในตัวอย่างนี้ type งานคือ worker และ index งานคือ 0
เพื่อจุดประสงค์ในการอธิบายประกอบ บทช่วยสอนนี้จะแสดงวิธีตั้งค่า TF_CONFIG กับ 2 คนทำงานบน localhost ในทางปฏิบัติ คุณจะต้องสร้างผู้ปฏิบัติงานหลายคนบนที่อยู่ IP ภายนอกและพอร์ต และตั้งค่า TF_CONFIG กับผู้ปฏิบัติงานแต่ละคนอย่างเหมาะสม เช่น แก้ไข index งาน
os.environ['TF_CONFIG'] = json.dumps({
'cluster': {
'worker': ["localhost:12345", "localhost:23456"]
},
'task': {'type': 'worker', 'index': 0}
})
กำหนดรูปแบบ
เขียนเลเยอร์ เครื่องมือเพิ่มประสิทธิภาพ และฟังก์ชันการสูญเสียสำหรับการฝึก บทช่วยสอนนี้กำหนดโมเดลด้วยเลเยอร์ Keras ซึ่งคล้ายกับบทช่วย สอนการฝึกอบรม GPU หลายตัว
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()))
MultiWorkerMirroredStrategy
ในการฝึกโมเดล ให้ใช้อินสแตนซ์ของ tf.distribute.experimental.MultiWorkerMirroredStrategy MultiWorkerMirroredStrategy สร้างสำเนาของตัวแปรทั้งหมดในเลเยอร์ของโมเดลบนอุปกรณ์แต่ละเครื่องของผู้ปฏิบัติงานทุกคน มันใช้ CollectiveOps ซึ่งเป็น TensorFlow op สำหรับการสื่อสารแบบกลุ่ม เพื่อรวมการไล่ระดับสีและทำให้ตัวแปรซิงค์กัน คู่มือ 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ใช้กลุ่มตามวงแหวนโดยใช้ gRPC เป็นเลเยอร์การสื่อสารข้ามโฮสต์ -
NCCLใช้ NCCL ของ Nvidia เพื่อปรับใช้ส่วนรวม -
AUTOเลื่อนตัวเลือกไปที่รันไทม์
ทางเลือกที่ดีที่สุดของการใช้งานแบบรวมขึ้นอยู่กับจำนวนและประเภทของ GPU และการเชื่อมต่อเครือข่ายในคลัสเตอร์ หากต้องการลบล้างตัวเลือกอัตโนมัติ ให้ระบุค่าที่ถูกต้องให้กับพารามิเตอร์
communicationของ Constructor ของMultiWorkerMirroredStrategyเช่นcommunication=tf.distribute.experimental.CollectiveCommunication.NCCL-
เยี่ยมชม ส่วนประสิทธิภาพ ในคู่มือเพื่อเรียนรู้เพิ่มเติมเกี่ยวกับกลยุทธ์และ เครื่องมือ อื่นๆ ที่คุณสามารถใช้เพื่อเพิ่มประสิทธิภาพการทำงานของแบบจำลอง TensorFlow ของคุณ
ตัวอย่างโค้ดอื่นๆ
- ตัวอย่าง แบบครบวงจรสำหรับการฝึกอบรมผู้ปฏิบัติงานหลายคนในเทนเซอร์โฟลว์/ระบบนิเวศโดยใช้เทมเพลต Kubernetes ตัวอย่างนี้เริ่มต้นด้วยโมเดล Keras และแปลงเป็นตัวประมาณโดยใช้
tf.keras.estimator.model_to_estimatorAPI - โมเดลอย่างเป็นทางการ ซึ่งหลายแบบสามารถกำหนดค่าให้รันกลยุทธ์การจัดจำหน่ายได้หลายแบบ