深度学习大数据预处理：构建高效数据流水线指南

原创

用户11764306

发布于 2025-11-04 20:23:48

790

ETL：提取、转换、加载

在数据库领域，ETL代表提取、转换、加载这三个数据流水线的基本构建块：

提取：从多个同构或异构源中提取数据
转换：指数据清洗和操作，将其转换为适当格式
加载：将转换后的数据注入处理单元内存中进行训练

构建这三个步骤的序列还不够，同样重要的是使它们快速运行。速度和性能是构建数据流水线的关键部分。

数据读取

数据读取或提取是从数据源获取数据并将其从存储格式转换为我们所需格式的步骤。在生产环境中，数据可能存储在数据库、对象存储云服务、数据仓库或本地简单存储单元中。每个存储选项都有自己的一套提取和解析数据规则。

从多个源加载

我们需要能够将所有不同的源合并和组合成可以传递到流水线下一步的单个实体。每个源都有特定的存储格式，因此我们还需要一种解码方法。

以下是使用tf.data的样板代码：

files = tf.data.Dataset.list_files(file_pattern)
dataset = tf.data.TFRecordDataset(files)

对于不支持的数据集，我们需要自己创建自定义数据加载器。

并行数据提取

在数据远程存储的情况下，加载它们可能成为流水线中的瓶颈。并行化是解决此瓶颈的答案：

tf.data.Dataset.list_files(file_pattern)
tf.data.interleave(TFRecordDataset, num_calls=4)

interleave()函数将同时加载许多数据点并交错结果，这样我们就不必等待每个数据点加载完成。

数据处理

在加载来自所有源的并行数据后，我们准备对它们应用一些转换。在此步骤中，我们运行计算最密集的函数，如图像操作、数据解码等。

函数式编程

函数式编程是一种编程范式，通过堆叠纯函数构建软件，避免在它们之间共享状态并使用不可变数据。

在函数式编程中，最重要的函数之一是map()。使用map()，我们可以应用几乎任何我们能想到的函数。

m_list = list(map(lambda i: i+1, m_list))

实际应用

在分割示例中，我们可以执行以下操作：

@staticmethod
def _preprocess_train(datapoint, image_size):
    """加载并预处理单个训练图像"""
    input_image = tf.image.resize(datapoint['image'], (image_size, image_size))
    input_mask = tf.image.resize(datapoint['segmentation_mask'], (image_size, image_size))

    if tf.random.uniform(()) > 0.5:
        input_image = tf.image.flip_left_right(input_image)
        input_mask = tf.image.flip_left_right(input_mask)

    input_image, input_mask = DataLoader._normalize(input_image, input_mask)
    return input_image, input_mask

我们可以使用map()和lambda函数简单地将预处理步骤添加到数据流水线中：

@staticmethod
def preprocess_data(dataset, batch_size, buffer_size, image_size):
    """预处理并拆分为训练和测试"""
    train = dataset['train'].map(lambda image: DataLoader._preprocess_train(image, image_size), 
                                num_parallel_calls=tf.data.experimental.AUTOTUNE)
    train_dataset = train.shuffle(buffer_size)

    test = dataset['test'].map(lambda image: DataLoader._preprocess_test(image, image_size))
    test_dataset = test.shuffle(buffer_size)

    return train_dataset, test_dataset

除了map()，tf.data还支持许多其他有用的函数：