将数据从`tf.data.Dataset`分发给多个工人(例如对Horovod)

Question

使用Horovod，您基本上可以运行N个独立的实例(因此它是图之间复制的一种形式)，它们通过特殊的Horovod操作(基本上是广播+减少)进行通信。

现在假设实例0或其他一些外部实例加载您的数据(通过tf.data.Dataset)。如何将iterator.get_next()分发给每个实例？使用Horovod广播将效率低下，因为您会将所有数据复制到所有实例。

在每个实例中拥有数据集，并在其中执行所有加载，然后在数据集上使用shard也是效率低下的，因为您将数据加载到任何地方，然后丢弃(N-1)数据集的/N。因此，这也是为什么不希望分片，而只是将数据集加载在单个(生产者/数据集工作者)实例中，然后将批处理分配到所有列车工人上。

我想TF MultiDeviceIterator提供了一些类似的功能(或者基本上就是这样)，但我不确定它是否与Horovod一起工作，以及您将如何设置它？

或者也许您可以通过TF员工(指南 )来进行分发？(也许这也是配置MultiDeviceIterator的方式？)

如果可能的话，这应该是通过TensorFlow操作/函数(有许多相关的函数可能已经给了我这一点，但我可能不知道，或者误解了它们的工作方式)。或者，答案可能是TensorFlow还没有提供任何这样的功能？(这将仍然是有用的了解。然后，我将在C++中实现自己的解决方案，包装为TensorFlow。但在这样做之前，最好知道这是否真的有必要。)

(霍洛沃德问题也与此相关。)

(这个问题实际上比Horovod更通用，尽管Horovod可能是一个很好的例子。对于分布式TensorFlow，您可能总是会遇到这个问题吗？)

(我收集了所有分布式TensorFlow术语和方面这里的概述，主要是为了澄清。)

(亲属关系是(也许？)还有这，这，这，这或这问题。)

Answer 1

正如您所说，在每个实例中复制数据并为每个实例分割数据是不切实际的。

然后，一种解决方案是将数据进程中的数据分开，让每个实例从数据进程中提取数据，如下图所示。例如，可以使用队列来建立这种通信。

在这样的系统中，数据进程将加载数据集，将其预处理为批处理，并将批处理推入队列。然后，每个训练实例将从这个队列中提取批。例如，可以将队列作为生成器传递到dataset API (请参阅生成器)。此外，如果批处理的生成速度不够快，则可以创建更多的数据进程来提高批处理吞吐量。

根据您的用例，实现细节将有所不同。有关更多信息，您可以查找联网和进程间通信和多处理管道和队列。

                                                             Training        
                                                         +--------------+  ++
                                                         |              |   |
                                                    +----+  Instance 1  |   |
                                                    |    |              |   |
                                                    |    +--------------+   |
                                                    |                       |
                      Preprocessing                 |                       |
                  +--------------------+            +---->      X           |
                  |                    |            |                       |
             Load |                    | Batches    +           X           |
    Dataset+------>    Data Process    +--------->Queue                     |  N instances
                  |                    |            +           X           |  Distributed training
                  |                    |            |                       |  For example, using
                  +--------------------+            +---->      X           |  Horovod broadcast + reduce
                                                    |                       |
                                                    |        Training       |
                                                    |    +--------------+   |
                                                    |    |              |   |
                                                    +----+  Instance N  |   |
                                                         |              |   |
                                                         +--------------+  ++

对于tensorflow实现，可以将tf.data.Dataset.shard与tf.data.TFRecordDataset结合使用。

这些文档使用TFRecords解决了您的低效率问题：

重要注意事项：

• 在使用任何随机运算符(例如洗牌)之前，一定要将其分解。
• 通常，最好是在dataset管道中早期使用shard操作符。例如，当读取一组TFRecord文件时，在将数据集转换为输入示例之前将其分解。这避免了读取每个工作人员上的每个文件。以下是完整管道内有效分割策略的示例：

D=Dataset.list_files(模式)d= d.shard(num_workers，worker_index) d= d.repeat(num_epochs) d= d.shuffle(shuffle_buffer_size) d= d.interleave(tf.data.TFRecordDataset，cycle_length=num_readers，block_length=1) d= d.map(parser_fn，num_parallel_calls=num_map_threads)

Answer 2

我记得看了一下YogaDL。它允许您缓存数据集，这样在培训(或再培训)期间，您将只访问该碎片上需要的数据，而不是丢弃(N-1)/N您的数据读取。