python - 如何使用 Python 中的线程来并行化 AWS S3 API 调用？

Question

我编写了一个 Python 脚本，旨在通过使用 AWS Boto 3 list_objects() method 来确定所有可用 AWS S3 存储桶的总大小。 .

逻辑很简单:

从每个 S3 存储桶中获取对象的初始列表(在 1,000 个对象后自动截断)

遍历对象列表中的每个对象，将该对象的大小添加到 total_size 变量

当存储桶仍有其他对象时，检索它们并重复步骤 2

这是相关的代码片段:

import boto3

s3_client = boto3.client('s3')

# Get all S3 buckets owned by the authenticated sender of the request
buckets = s3_client.list_buckets()

# For each bucket...
for bucket in buckets['Buckets']:
    # Get up to first 1,000 objects in bucket
    bucket_objects = s3_client.list_objects(Bucket=bucket['Name'])

    # Initialize total_size
    total_size = 0

    # Add size of each individual item in bucket to total size
    for obj in bucket_objects['Contents']:
        total_size += obj['Size']

    # Get additional objects from bucket, if more
    while bucket_objects['IsTruncated']:
        # Get next 1,000 objects, starting after final object of current list
        bucket_objects = s3_client.list_objects(
            Bucket=bucket['Name'],
            Marker=bucket_objects['Contents'][-1]['Key'])
        for obj in bucket_objects['Contents']:
            total_size += obj['Size']

    size_in_MB = total_size/1000000.0
    print('Total size of objects in bucket %s: %.2f MB'
        % (bucket['Name'], size_in_MB))

此代码在其中包含少于 5 MB 左右数据的存储桶上运行相对较快，但是当我遇到其中包含 90+ MB 数据的存储桶时，执行时间从毫秒跃升至 20-30+ 秒。

我希望使用 threading module并行化代码的 I/O 部分(从 S3 获取对象列表)，以便在线程检索完成后立即添加存储桶中所有对象的总大小，而不必进行检索和添加依次。

我知道由于 GIL，Python 不支持真正的多线程，只是为了避免对这种效果做出响应，但我的理解是，由于这是 I/O 操作而不是 CPU 密集型操作，线程模块应该能够提高运行时间。

我的问题与我在这里看到的线程实现的几个示例之间的主要区别在于，我没有迭代已知的列表或集合。在这里，我必须首先检索对象列表，查看列表是否被截断，然后根据当前列表中最终对象的键检索下一个对象列表。

任何人都可以解释一种改善此代码运行时间的方法，还是在这种情况下不可能？

Answer 1

我遇到了类似的问题。
为每个线程创建一个单独的 session 似乎很重要。
所以而不是

s3_client = boto3.client('s3')

你需要写

s3_client = boto3.session.Session().client('s3')

否则线程会相互干扰，并发生随机错误。
除此之外，多线程的正常问题也适用。
我的项目是将 135,000 个文件上传到 S3 存储桶。到目前为止，我发现使用 8 个线程可以获得最佳性能。否则需要 3.6 小时，需要 1.25 小时。