为啥 TensorFlow 的 `tf.data` 包会减慢我的代码速度？

Posted 2023-02-23

【中文标题】为啥 TensorFlow 的 `tf.data` 包会减慢我的代码速度？

【英文标题】：Why is TensorFlow's `tf.data` package slowing down my code?为什么 TensorFlow 的 `tf.data` 包会减慢我的代码速度？

【发布时间】：2019-01-03 14:43:39

【问题描述】：

我刚刚学习使用 TensorFlow 的 tf.data API，我发现它大大降低了我的代码速度，以每个 epoch 的时间来衡量。这与它应该做的相反，我想。我写了一个简单的线性回归程序来测试它。

Tl;Dr：使用 100,000 个训练数据，tf.data 将每个 epoch 的时间减慢大约 10 倍，如果您使用的是完整的批量训练。如果您使用较小的批次，情况会更糟。 500 个训练数据则相反。

我的问题：这是怎么回事？我的实施有缺陷吗？我读过的其他来源有 tf.data 将速度提高了大约 30%。

import tensorflow as tf 
import numpy as np
import timeit

import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
tf.logging.set_verbosity(tf.logging.ERROR)

n_epochs = 10
input_dimensions_list = [10]

def function_to_approximate(x):
    return np.dot(x, random_covector).astype(np.float32) + np.float32(.01) * np.random.randn(1,1).astype(np.float32)

def regress_without_tfData(n_epochs, input_dimension, training_inputs, training_labels):
    tf.reset_default_graph()
    weights = tf.get_variable("weights", initializer=np.random.randn(input_dimension, 1).astype(np.float32))

    X = tf.placeholder(tf.float32, shape=(None, input_dimension), name='X')
    Y = tf.placeholder(tf.float32, shape=(None, 1), name='Y')
    prediction = tf.matmul(X,weights)
    loss = tf.reduce_mean(tf.square(tf.subtract(prediction, Y)))
    loss_op = tf.train.AdamOptimizer(.01).minimize(loss)

    init = tf.global_variables_initializer()

    with tf.Session() as sess:
        sess.run(init)
        for _ in range(n_epochs):
            sess.run(loss_op, feed_dict=X: training_inputs, Y:training_labels)

def regress_with_tfData(n_epochs, input_dimension, training_inputs, training_labels, batch_size):
    tf.reset_default_graph()
    weights = tf.get_variable("weights", initializer=np.random.randn(input_dimension, 1).astype(np.float32))

    X,Y = data_set.make_one_shot_iterator().get_next()

    prediction = tf.matmul(X, weights)
    loss = tf.reduce_mean(tf.square(tf.subtract(prediction, Y)))
    loss_op = tf.train.AdamOptimizer(.01).minimize(loss)

    init = tf.global_variables_initializer()

    with tf.Session() as sess:
        sess.run(init)
        while True:
            try: 
                sess.run(loss_op)
            except tf.errors.OutOfRangeError:
                break

for input_dimension in input_dimensions_list:
    for data_size in [500, 100000]:

        training_inputs = np.random.randn(data_size, input_dimension).astype(np.float32)
        random_covector = np.random.randint(-5, 5, size=(input_dimension, 1))
        training_labels = function_to_approximate(training_inputs)

        print("Not using tf.data, with data size "
        ", input dimension  and training with "
        "a full batch, it took an average of "
        " seconds to run  epochs.\n".
            format(
                data_size,
                input_dimension,
                timeit.timeit(
                    lambda: regress_without_tfData(
                        n_epochs, input_dimension, 
                        training_inputs, training_labels
                    ), 
                    number=3
                ),
                n_epochs))

for input_dimension in input_dimensions_list:
    for data_size, batch_size in [(500, 50), (500, 500), (100000, 50), (100000, 100000)]:

        training_inputs = np.random.randn(data_size, input_dimension).astype(np.float32)
        random_covector = np.random.randint(-5, 5, size=(input_dimension, 1))
        training_labels = function_to_approximate(training_inputs)

        data_set = tf.data.Dataset.from_tensor_slices((training_inputs, training_labels))
        data_set = data_set.repeat(n_epochs)
        data_set = data_set.batch(batch_size)

        print("Using tf.data, with data size "
        ", and input dimension , and training with "
        "batch size , it took an average of  seconds "
        "to run  epochs.\n".
            format(
                data_size,
                input_dimension,
                batch_size,
                timeit.timeit(
                    lambda: regress_with_tfData(
                        n_epochs, input_dimension, 
                        training_inputs, training_labels, 
                        batch_size
                    ),
                    number=3
                )/3,
                n_epochs
            ))

这为我输出：

不使用tf.data，数据大小500，输入维度10，训练 一个完整的批次，平均需要 0.20243382899980134 秒 运行 10 个 epoch。

不使用 tf.data，数据大小 100000，输入维度 10 和 全批次训练，平均耗时 0.2431719040000644 运行 10 个 epoch 的秒数。

使用 tf.data，数据大小为 500，输入维度为 10，以及 批量大小为 50 的训练，平均需要 0.09512088866661846 运行 10 个 epoch 的秒数。

使用 tf.data，数据大小为 500，输入维度为 10，以及 批量大小为 500 的训练，平均需要 0.07286913600000844 秒运行 10 个 epoch。

使用 tf.data，数据大小为 100000，输入维度为 10，以及 批量大小为 50 的训练，平均需要 4.421892363666605 运行 10 个 epoch 的秒数。

使用 tf.data，数据大小为 100000，输入维度为 10，以及 批量大小为 100000 的训练，平均需要 2.2555197536667038 秒运行 10 个 epoch。

编辑：修复了 Fred Guth 指出的一个重要问题。不过对结果影响不大。

【问题讨论】：

为什么每次通话都定义数据集？

【参考方案1】：

接受的答案不再有效，因为 TF 行为已经改变。根据文档：

from_tensors 生成一个仅包含单个元素的数据集。到 将输入张量切成多个元素，使用 from_tensor_slices 而是。

这意味着你不能批处理它

X = np.arange(10)
data = tf.data.Dataset.from_tensors( X )
data = data.batch(2)
for t in data.as_numpy_iterator():
  print(t)
# only one row, whereas expected 5 !!!

文档推荐from_tensor_slices。但是与 numpy 切片相比，这有相当多的开销。慢切片是一个悬而未决的问题https://github.com/tensorflow/tensorflow/issues/39750

本质上，TF 中的切片速度很慢，并且会影响输入绑定或轻型模型，例如小型网络（回归、word2vec）。

【参考方案2】：

那是因为您将苹果与香蕉进行比较。

一方面，当使用占位符时，您提供的是一个整体张量。另一方面，当使用Dataset 时，您将张量分割成单个样本。这是非常不同的。

使用Dataset 管道提供单片占位符张量的等价物是使用tf.data.Dataset.from_tensors。 当我在您的示例中使用 from_tensors 时，我得到的计算时间与使用占位符相似（实际上更短）。

如果您想使用from_tensor_slices 比较更复杂的管道，您应该使用占位符进行公平比较。例如，随机播放您的数据。在切片上添加一些预处理。我毫不怀疑您会观察到使人们转向此管道的性能提升。

【讨论】：

当您使用from_tensors 获得较低次数时，您使用了哪种迭代器？例如，您调用了repeat 方法的一次性操作？还是一个可初始化的，你每次在运行训练操作之前都重新初始化？

我使用了一次性迭代器。您在重现结果时遇到问题吗？

你打电话给my_dataset = my_dataset.repeat()?

【参考方案3】：

您可能缺少的一件事是预取。在数据管道的末尾添加 1 的预取，如下所示：

data_set = tf.data.Dataset.from_tensor_slices((training_inputs, training_labels))
data_set = data_set.repeat(n_epochs)
data_set = data_set.batch(batch_size).prefetch(1)

在数据集管道的末尾添加 1 的预取意味着您在进行训练时尝试获取 1 批数据。这样您就不会在准备批次时等待，它应该在每次训练迭代完成后立即准备就绪。

【参考方案4】：

我想测试似乎非常方便处理数据的数据集 API。我在 CPU、GPU 和多 GPU 方式中针对具有不同数据类型的小型和大型 NN 对这个 API 做了很多时间测试。

首先，在我看来，您的代码没问题。但我需要指出，你的神经网络只是一个简单的层。

现在，数据集 API 不适合您的 NN 类型，但适合复杂得多的 NN。为什么 ？出于我在下面解释的几个原因（基于我对理解数据集 API 的追求）。

首先，数据集 API 每批处理数据，另一方面数据经过预处理。因此，如果它适合您的 RAM，您可以通过预处理数据来节省时间。在这里，您的数据只是为了“简单”。如果你想测试我在说什么，试着找到一个非常大的数据集来处理。不过，可以使用 prefetching 数据调整数据集 API。你可以看看这个tutorial，它很好地解释了为什么使用预取来处理数据是好的。

其次，在我寻求用于多 GPU 训练的数据集 API 时，我发现据我所知 旧的预处理方式比用于小型神经网络的数据集 API 更快。您可以通过创建一个简单的可堆叠 RNN 来验证这一点，该 RNN 采用输入序列。您可以尝试不同大小的堆栈（我测试了 1、2、10 和 20）。您会看到，使用数据集 API，在 1-GPU 或 4-GPU 上，小型 RNN 堆栈（1、2 和 5）的时间没有差异。

总而言之，数据集 API 适用于无法预处理数据的神经网络。根据您的任务，预处理数据可能更方便，例如，如果您想调整您的 NN 以改进它。我同意数据集 API 对于批处理、填充和混洗大量数据非常方便，但它也不适合多 GPU 训练。

【参考方案5】：

第一：

您正在不必要地重新创建数据集。

data_set = tf.data.Dataset.from_tensor_slices((training_inputs, training_labels))

在循环之前创建数据集并将regress_with_tfData 输入签名更改为使用数据集而不是training_inputs 和training_labels。

第二：

这里的问题是，大小为 50 甚至 500 的 minibatch 太小，无法补偿 td.data 构建延迟的成本。您应该增加小批量大小。有趣的是，您使用大小为 100000 的 minibatch 执行此操作，但可能它太大了（我不确定，我认为它需要更多测试）。

您可以尝试以下几种方法：

1) 将 minibatch 大小增加到 10000 左右，看看是否有改进 2) 更改您的管道以使用迭代器，例如：

    data_set = tf.data.Dataset.from_tensor_slices((training_inputs, training_labels))
    data_set = data_set.repeat(n_epochs)
    data_set = data_set.batch(batch_size)
    iterator = data_set.make_one_shot_iterator()
    ....
    next_element = iterator.get_next()

【讨论】：

谢谢。我同意这是一个错误。但我不同意它在每次迭代中都被重新定义……只是每次我想运行一个新模型时。 （每次我有一个新模型时，我只调用一次regress_with_tfData。）而且我认为每次运行模型时都应该计算时间（即使我可以绕过它），因为它是影响模型运行时间长短的一个因素。模型需要。无论如何，我将它从函数中取出，不幸的是，问题仍然存在。

很公平，我同意你的看法。可以标记为已解决吗？我编辑了答案。

我的管道不是已经是迭代器了吗？我看不出你的示例代码和我写的有什么区别。

你是对的......再次 :grin: 我想念data_set.make_one_shot_iterator().get_next()

注意 make_one_shot_iterator() 在 TF2.0 中已被弃用，您只需使用 iterator = iter(data_set)