tf.distribute.MirroredStrategy需在scope内构建模型、编译及数据集,训练时用strategy.run()并行执行、strategy.reduce()同步梯度;MultiWorkerMirroredStrategy依赖TF_CONFIG和网络配置;数据加载应合理使用cache、map和prefetch。

TensorFlow 2.x 中 tf.distribute.MirroredStrategy 是最常用的数据并行方案
它适合单机多 GPU 场景,自动把模型和计算图复制到每个 GPU 上,同步更新梯度。不是所有模型都能直接套用——必须在策略作用域内创建模型、优化器和数据集,否则会报 NotImplementedError: Cannot copy tensor from device 或梯度为空。
常见错误是把 model.compile() 放在 MirroredStrategy().scope() 外面,或者用普通 tf.data.Dataset 而没调用 strategy.experimental_distribute_dataset()。
- 必须在
strategy.scope()内构建模型、编译、定义损失和优化器 - 训练数据需用
strategy.experimental_distribute_dataset(dataset)包装,不能直接传给model.fit() - 自定义训练循环时,要用
strategy.run()包裹单步训练逻辑,并用strategy.reduce()汇总跨设备结果
自定义训练循环中如何正确使用 strategy.run() 和 strategy.reduce()
用 model.fit() 看似省事,但一旦需要梯度裁剪、多损失加权或动态学习率,就得切到自定义循环。这时容易漏掉设备间同步逻辑,导致各 GPU 算出的 loss 不一致、梯度未平均就更新。
strategy.run() 在每个设备上并行执行函数,返回一个 DistributedValues 对象;而 strategy.reduce() 才真正做均值(tf.distribute.ReduceOp.SUM 或 tf.distribute.ReduceOp.MEAN)并拉回 host。
立即学习“Python免费学习笔记(深入)”;
- 输入 batch 必须已由
experimental_distribute_dataset分发,否则strategy.run()接收的是 host 侧张量,无法分发 -
strategy.reduce()的value参数必须是strategy.run()的返回值,不能是中间变量(如某 GPU 上的局部 loss) - 优化器的
apply_gradients()必须在strategy.run()内部调用,否则梯度不会被同步
@tf.function
def train_step(inputs):
features, labels = inputs
with tf.GradientTape() as tape:
predictions = model(features, training=True)
loss = loss_fn(labels, predictions)
gradients = tape.gradient(loss, model.trainable_variables)
# 注意:apply_gradients 必须在 run 内部
optimizer.apply_gradients(zip(gradients, model.trainable_variables))
return loss
<h1>正确调用方式</h1><p>per_replica_loss = strategy.run(train_step, args=(dist_inputs,))
loss = strategy.reduce(tf.distribute.ReduceOp.MEAN, per_replica_loss, axis=None)</p>为什么 tf.distribute.MultiWorkerMirroredStrategy 容易卡在初始化阶段
多机训练不是简单把单机代码加个 strategy 就行。它依赖 gRPC 通信,要求所有 worker 使用相同 Python 环境、TF 版本、CUDA/cuDNN 版本,并且网络互通(特别是端口 12345 默认用于协调器)。最常见的失败是 FailedPreconditionError: Worker failed to connect to coordinator 或长时间无响应。
关键点在于环境变量设置和任务角色划分:必须显式指定 TF_CONFIG,且 coordinator(chief)节点要先启动,其他 worker 等待其就绪。
-
TF_CONFIG是 JSON 字符串,必须包含cluster(含worker列表)和task(type和index) - 所有 worker 启动命令需一致,仅
TF_CONFIG中的task.index不同 - 推荐用
os.environ['TF_CONFIG'] = json.dumps(...)设置,不要用 shell export,Python 子进程可能读不到
数据并行下 tf.data.Dataset 的性能陷阱
数据加载常成瓶颈,尤其当 batch_size 增大后,GPU 等待数据时间变长。很多人以为加了 prefetch(tf.data.AUTOTUNE) 就够了,但忽略了 num_parallel_calls 和 cache() 的配合时机。
典型低效写法是:先 map() 解码图片,再 cache() —— 这会让缓存内容是原始字节流,每次仍要解码;或者 prefetch() 放在 batch() 之后,失去重叠 IO 与计算的意义。
- 图像类任务建议:
cache()放在map()解码之后、batch()之前;文本类若内存够,也可缓存 tokenized 结果 -
map(..., num_parallel_calls=tf.data.AUTOTUNE)必须设,否则默认为 1,CPU 核心基本闲置 -
prefetch()一定要放在流水线最后,即dataset.prefetch(tf.data.AUTOTUNE),且不能少于 1
分布式下还要注意:每个 worker 加载的是全量数据的子集(shard),所以 shuffle(buffer_size) 的 buffer_size 应设为单 worker 数据量的 1–2 倍,而非全局总量,否则 shuffle 效果差。


















