1TB 数据集上训练 XGBoost:SageMaker 并行分布式训练数据

随着机器学习的不断发展，我们看到具有越来越多参数的更大模型。与此同时，我们也看到了令人难以置信的大型数据集，归根结底，任何模型的好坏都取决于它所训练的数据。使用大型模型和数据集的计算成本可能很高，并且难以及时迭代或试验。在本文中，我们将重点介绍问题的大型数据集部分。具体来说，我们将研究一种称为分布式数据并行的东西，利用来优化和减少大型真实数据集中的训练时间。

对于今天的示例，我们将在人工生成的 1 TB 数据集上训练算法。在此示例中，我们将更深入地了解如何准备和构建数据源以加快训练速度，以及如何使用内置的数据并行库启动分布式训练。

注意：本文将假定 AWS 和特定子功能的基本知识，例如培训以及通过 SDK 和 Boto3 AWS SDK 与和 AWS 进行整体交互。有关培训的正确介绍和概述，我将参考本文。

什么是分布式数据并行？我们为什么需要它？

在我们开始实施之前，了解分布式数据训练至关重要。对于大型数据集，优化训练时间确实很困难，因为这是非常计算密集型的。您必须考虑能够将数据集下载到内存中以及机器必须执行的任何训练和超参数计算。使用一台机器，这是可能的（如果计算能力足够强大），但从时间的角度来看效率低下，实验成为一场噩梦。

使用分布式数据并行，您可以使用实例集群。其中每个实例都可以包含多个 CPU/GPU。从头开始创建此分布式数据并行设置可能具有挑战性，并且需要解决节点到节点通信的大量开销。为了简化问题，我们可以利用内置的分布式数据并行库。在这里，构建和优化节点到节点通信的艰苦工作被抽象出来，您可以专注于模型开发。

为什么数据源对很重要？

提供数据的位置和方式对于优化训练时间至关重要。使用，事实上的存储服务一直是S3，这仍然是一个选项。有一种普通训练模式，您可以将数据集直接上传到 S3，这称为文件模式。在这里，在训练开始之前将您的数据集下载到实例内存中。对于今天的示例，我们将使用称为快速文件模式的优化 S3 模式。使用快速文件模式，数据集实时流式传输到实例中，因此我们可以避免下载整个数据集的开销。这也导致了一个问题，我应该如何提供/拆分我的数据集？对于此示例，我们将数据集拆分为多个较小的文件，总计 1TB，这将再次有助于下载，或者在我们的情况式传输时间以及我们的数据集规模非常大。

在 S3 之外，还可以在上使用弹性文件系统（EFS）和 FsX 。如果您的训练数据已驻留在 EFS 上，则可以轻松挂载到上。与其他选项相比，使用 FsX ，您可以以更高的速度进行扩展，但为此选项设置 VPC 会产生运营开销。

归根结底，在使用的数据源的不同训练选项之间做出决定时，应考虑许多因素。要考虑的两个要点是数据集大小以及如何对数据集进行分片，这些因素与数据集的当前位置相结合，将帮助您做出正确的选择来优化训练时间。有关更全面的指南，请参阅这篇有关培训数据源的文章。

数据集创建

在本例中，我们将利用鲍鱼数据集，并对其运行算法作为回归模型。您可以从公开可用的数据集下载数据集。

#retreive data
aws s3 cp s3://sagemaker-sample-files/datasets/tabular/uci_abalone/train_csv/abalone_dataset1_train.csv .

这个数据集本身只有 100KB，所以我们需要制作它的大量副本来创建 1TB 的数据集。对于此数据集准备，我利用 EC2 实例（r6a.）进行开发。这是一个高内存和计算实例，允许快速准备我们的数据集。设置完成后，我们运行以下脚本，将数据集转换为更大的 100MB 文件。您可以根据需要拼接实际数据，这不是需要遵循的固定配方/大小。

import os
import pandas as pd
import sys
#~110KB initial file
df = pd.read_csv("abalone_dataset1_train.csv")
print(sys.getsizeof(df))
#creates a 104MB file
df_larger = pd.concat([df]*700, ignore_index=True)
print(sys.getsizeof(df_larger))
df_larger.to_csv("abalone-100mb.csv")

使用 100MB 数据集，我们可以制作 10，000 个副本来创建我们的 1TB 数据集。然后，我们将这 10，000 个副本上传到 S3，这是我们的快速文件模式数据源。

%%sh
#replace with your S3 bucket to upload to
s3_bucket='sagemaker-us-east-1-474422712127'
for i in {0..10000}
do
  aws s3 cp abalone-100mb.csv s3://$s3_bucket/xgboost-1TB/abalone-$i.csv 
done

此脚本大约需要 2 小时才能运行，但如果您想加快操作速度，您可以使用某种形式的多处理代码和 Boto3 来加快上传时间。

培训设置

在开始运行训练作业之前，我们需要设置正确的客户端和配置。在这里，我们专门定义了训练实例类型，您可以在以下页面找到广泛的选项列表。在选择实例类型时，您需要考虑您正在处理的模型类型和您所在的域。对于 NLP 和计算机视觉用例，通常 GPU 实例已被证明更适合，在这种情况下，我们使用内存优化实例和算法。

import boto3
import sagemaker
from sagemaker.estimator import Estimator
boto_session = boto3.session.Session()
region = boto_session.region_name
sagemaker_session = sagemaker.Session()
base_job_prefix = 'xgboost-example'
role = sagemaker.get_execution_role()
default_bucket = sagemaker_session.default_bucket()
s3_prefix = base_job_prefix
training_instance_type = 'ml.m5.24xlarge'

接下来我们准备我们的训练输入，在这里我们指定我们正在使用快速文件模式，否则它默认为文件模式。我们还将分配指定为“”，这表明我们希望在所有实例中分发所有不同的 S3 文件。否则，所有数据文件都将加载到每个实例中，从而导致更长的训练时间。

from sagemaker.inputs import TrainingInput
#replace with your S3 Bucket with data
training_path = 's3://sagemaker-us-east-1-474422712127/xgboost-1TB/'
#set distribution to ShardedByS3Key otherwise a copy of all files will be made across all instances
#we also enable FastFile mode here where as the default is File mode
train_input = TrainingInput(training_path, content_type="text/csv", 
input_mode='FastFile', distribution = "ShardedByS3Key")
training_path

然后，我们准备了估计器，以更深入地了解算法及其可用的超参数，请参阅本文。这里的另一个关键是我们将实例计数指定为 25（请注意，您可能需要根据实例在此处请求提高限制）。我们的 10，000 个数据文件将分布在这 25 ml.m5. 实例中。一旦我们指定一个大于的计数，就会为我们的模型推断分布式数据并行。

model_path = f's3://{default_bucket}/{s3_prefix}/xgb_model'
image_uri = sagemaker.image_uris.retrieve(
    framework="xgboost",
    region=region,
    version="1.0-1",
    py_version="py3",
    instance_type=training_instance_type,
)
xgb_train = Estimator(
    image_uri=image_uri,
    instance_type=training_instance_type,
    instance_count=25,
    output_path=model_path,
    sagemaker_session=sagemaker_session,
    role=role,
    
)
xgb_train.set_hyperparameters(
    objective="reg:linear",
    num_round=50,
    max_depth=5,
    eta=0.2,
    gamma=4,
    min_child_weight=6,
    subsample=0.7,
    silent=0,
)
training_instance_type

然后，我们可以通过在训练输入上拟合算法来启动训练作业。

xgb_train.fit({'train': train_input})

使用我们当前的设置，此训练作业大约需要 11 小时才能完成。

如何进一步优化？

我们可以通过几种不同的方式调整此训练时间。一种选择是通过增加实例计数来简单地水平扩展。另一种选择是采用 GPU 实例路由，这可能需要较少的实例计数，但这并不总是一门直接的科学。

除了调整训练作业背后的硬件之外，我们还可以重新访问我们所说的数据源格式。您可以评估 FsX ，它可以扩展到 100 GB/s 的吞吐量。另一种选择是以不同的格式对数据集进行分片，以尝试文件数量和文件大小的各种组合。