如何使用Boto3来遍历 AWS Glue 中存在的 S3 存储桶中的所有对象

AWS Glue是一个完全托管的数据集成服务，用于使数据准备和加载变得更加容易和便捷。AmazonS3存储桶是一个非常受欢迎的存储提供程序，经常用于存储各种类型的数据，包括图像，音频和视频文件。

在AWS Glue中，可以使用Python编写ETL作业。其中，Python客户端库Boto3提供了很多功能，可以用来实现与AWS资源的交互。在这篇文章中，我们将会学习如何使用Boto3来遍历在AWS Glue中存在的S3存储桶中的所有对象。

阅读更多：Python 教程

准备工作

在开始本文的实现步骤之前，您需要满足以下几个前提条件：

拥有一个AWS账户
已经创建了一个AWS Glue Job，并将Python脚本上传到了AWS S3存储桶中
拥有一个AWS S3存储桶

前提条件都准备好后，我们开始编写 Python 代码。

编写Python代码

首先，您需要导入AWS Python SDK的Boto3库。在Python代码中，只需要运行以下命令。

import boto3

导入Boto3库后，我们需要在Amazon S3中指定bucket_name和prefix。

s3 = boto3.client('s3')
bucket_name = '<your_bucket_name>'
prefix = '<your_prefix: (optional)>'

接下来，我们就可以使用list_objects_v2()方法来获取存储桶中所有对象的清单。请参见以下示例代码：

def get_matching_s3_objects(bucket, prefix="", suffix=""):
    """
    Generate objects in an S3 bucket.

    :param bucket: Name of the S3 bucket.
    :param prefix: Only fetch objects whose key starts with
        this prefix (optional).
    :param suffix: Only fetch objects whose keys end with
        this suffix (optional).
    """
    kwargs = {'Bucket': bucket}
    if isinstance(prefix, str):
        kwargs['Prefix'] = prefix
    while True:
        resp = s3.list_objects_v2(**kwargs)
        for obj in resp['Contents']:
            key = obj['Key']
            if key.startswith(prefix) and key.endswith(suffix):
                yield obj
        try:
            kwargs['ContinuationToken'] = resp['NextContinuationToken']
        except KeyError:
            break

接下来，我们可以遍历存储桶中的所有对象。

for obj in get_matching_s3_objects(bucket_name, prefix):
    s3.download_file(bucket_name, obj['Key'], '<local_file_path>')

以上代码会下载所有文件到指定路径。如果存储桶中文件过多，建议使用AWS S3的流式下载。

流式下载S3对象

如果存储桶中的文件过多，遍历所有的对象，并下载到本地存储，可能会导致内存不足的问题。因此，我们可以使用AWS S3的分段下载机制，以流式方式下载和处理S3对象。

下面的代码演示了如何使用stream方式来下载S3对象，并进行数据处理。

def select_key(obj):
    return obj.key


bucket = s3.Bucket(bucket_name)
keys = [select_key(obj) for obj in bucket.objects.filter(Prefix=prefix)]

s3objs = bucket.objects.filter(Prefix=prefix)
for i, obj in enumerate(s3objs):
    obj = s3.Object(obj.bucket_name, obj.key)
    with io.BytesIO(obj.get()["Body"].read()) as file:
        file.seek(0)
        # 处理下载的对象