将大熊猫 DataFrame 保存到 S3 的最快方法是什么？

Question

我正在尝试找出将 LARGE pandas DataFrame 写入 S3 文件系统的最快方法。我目前正在尝试两种方法：

1）通过gzip压缩（BytesIO）和boto3

gz_buffer = BytesIO()

with gzip.GzipFile(mode='w', fileobj=gz_buffer) as gz_file:
    df.to_csv(TextIOWrapper(gz_file, 'utf8'), index=False)

s3_resource = boto3.resource('s3')
s3_object = s3_resource.Object(bucket, s3_path + name_zip)
s3_object.put(Body=gz_buffer.getvalue())

对于 7M 行的数据帧，写入 S3 大约需要 420 秒。

2) 通过写入 csv 文件而不压缩（StringIO 缓冲区）

csv_buffer = StringIO()
data.to_csv(csv_buffer)
s3_resource = boto3.resource('s3')
s3_resource.Object(bucket, s3_path + name_csv).put(Body=csv_buffer.getvalue())

大约需要 371 秒...

问题是： 还有其他更快的方法可以将 pandas 数据帧写入 S3 吗？

Answer 1

使用分段上传可以更快地传输到 S3。压缩使文件更小，这也会有所帮助。

import boto3
s3 = boto3.client('s3')

csv_buffer = BytesIO()
df.to_csv(csv_buffer, compression='gzip')

# multipart upload
# use boto3.s3.transfer.TransferConfig if you need to tune part size or other settings
s3.upload_fileobj(csv_buffer, bucket, key)

s3.upload_fileobj 的文档在这里：https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.upload_fileobj

Answer 2

您可以尝试使用 s3fs 和 pandas 压缩来上传到 S3。 StringIO 或 BytesIO 占用内存。

import s3fs
import pandas as pd

s3 = s3fs.S3FileSystem(anon=False)
df = pd.read_csv("some_large_file")
with s3.open('s3://bucket/file.csv.gzip','w') as f:
    df.to_csv(f, compression='gzip')

Answer 3

这真的取决于内容，但这与boto3 无关。首先尝试将您的 DataFrame 转储到本地，看看什么是最快的以及您获得的大小。

以下是我们发现速度很快的一些建议，适用于几 MB 到超过 2GB 的情况（尽管对于超过 2GB 的情况，您确实需要 parquet 并可能将其拆分为 parquet 数据集）：

1. 大量混合文本/数字数据（面向 SQL 的内容）：使用 df.to_parquet(file)。

2. 主要是数字数据（例如，如果您的列 df.dtypes 表示一个快乐的 numpy 单一类型数组，而不是 Object）：您可以尝试 df_to_hdf(file, 'key')。

一点建议：尝试将您的 df 拆分为一些对您有意义的碎片（例如，按时间序列）。特别是如果您对单个分片有大量更新（例如时间序列中的最后一个），它将使您的下载/上传速度更快。

我们发现，HDF5 体积更大（未压缩），但它们在内存中保存/加载的速度非常快。 Parquets 默认是快速压缩的，因此它们往往更小（当然，这取决于数据的熵；如果您保存完全随机数，则会受到惩罚）。

对于boto3 客户端，multipart_chunksize 和multipart_threshold 默认都是 8MB，这通常是一个不错的选择。您可以通过以下方式查看：

tc = boto3.s3.transfer.TransferConfig()
print(f'chunksize: {tc.multipart_chunksize}, threshold: {tc.multipart_threshold}')

此外，默认情况下每次上传使用 10 个线程（除非您的对象大小大于上述阈值，否则不会执行任何操作）。

另一个问题是如何有效地上传许多文件。 TransferConfig 中的任何定义不处理。但我跑题了，最初的问题是关于单个对象的。

Answer 4

首先，检查您写入的存储桶是否与您的笔记本位于同一区域。

其次，您可以尝试使用 multi-part 上传选项，该选项将大于几 GB 的文件并行上传：

from boto3.s3.transfer import TransferConfig

def s3_upload_file(args):     
    s3 = boto3.resource('s3')

    GB = 1024 ** 3
    config = TransferConfig(multipart_threshold=5 * GB)

    s3.meta.client.upload_file(args.path, args.bucket, os.path.basename(args.path),Config=config)