S3 boto3功能优化

Question

我正在使用 boto3，并且我有一个函数，它采用 s3 文件的最后修改时间（函数参数）并获取所有文件，整个最后修改时间是 = 函数参数或大于该函数参数。

这是我的函数，它也可以正常工作，但是由于 s3 存储桶中有数百万个文件，因此执行需要很长时间。无论如何，我们是否可以优化使用 boto3 从 s3 获取和过滤数据的方式？

def get_data_from_s3_time_basis(self, last_modified):
        s3_bucket= self.s3.Bucket(self.S3_BUCKET)
        prefix = f"{self.s3_folder}/{self.s3_schema}/{self.sub_folder}/"

        logger.info(f"Fetching s3 files from {prefix} >= than timestamp {last_modified}")

        data_at_last_modified = []
        data_greater_than_last_modified = []

        for file in s3_bucket.objects.filter(Prefix=prefix):
            file_name = file.key.replace(prefix, '')

            if file.last_modified.replace(tzinfo=datetime.timezone.utc) == last_modified.replace(
                    tzinfo=datetime.timezone.utc) and file_name != '':

                files_at_same_timestamp_row = (file_name, file.last_modified.replace(
                    tzinfo=datetime.timezone.utc))

                data_at_last_modified.append(files_at_same_timestamp_row)

            if file.last_modified.replace(tzinfo=datetime.timezone.utc) > last_modified.replace(
                    tzinfo=datetime.timezone.utc) and file_name != '':

                data_greater_than_last_modified_row = (file_name, file.last_modified.replace(
                    tzinfo=datetime.timezone.utc))

                data_greater_than_last_modified.append(data_greater_than_last_modified_row)

        return data_at_last_modified, data_greater_than_last_modified, prefix

到目前为止尝试过多线程：

    def get_data_from_s3_time_basis_async(self, s3_bucket, prefix, data_at_last_modified, data_greater_than_last_modified, last_processed_time):
        for s3object in s3_bucket.objects.filter(Prefix=prefix):
            s3file = s3object.key.replace(prefix, '')
            s3_last_modified = s3object.last_modified.replace(tzinfo=datetime.timezone.utc)
            if s3_last_modified == last_processed_time and s3file != '':
                files_at_same_timestamp_row = (s3file, s3_last_modified)
                data_at_last_modified.append(files_at_same_timestamp_row)

            if s3_last_modified > last_processed_time and s3file != '':
                data_greater_than_last_modified_row = (s3file, s3_last_modified)
                data_greater_than_last_modified.append(data_greater_than_last_modified_row)

        return data_at_last_modified, data_greater_than_last_modified, prefix

    def get_data_from_s3_time_basis(self, last_processed_file_timestamp):
        s3_bucket = self.s3.Bucket(self.S3_BUCKET)
        prefix = f"{self.s3_folder}/{self.s3_schema}/{self.sub_folder}/"
        last_processed_time = last_processed_file_timestamp.replace(tzinfo=datetime.timezone.utc)

        data_at_last_modified = []
        data_greater_than_last_modified = []

        with concurrent.futures.ThreadPoolExecutor(max_workers=1000) as executor:
            executor.map(self.get_data_from_s3_time_basis_async, s3_bucket, prefix, data_at_last_modified, data_greater_than_last_modified, last_processed_time)

Answer 1

一种有效的方法是检索所有文件的 URL，然后通过多线程或异步一次对多个文件并行执行所需的操作。 可以在here 找到关于 python 并行处理的一个很好的介绍。 如果您更愿意使用异步 IO（我推荐您的用例），有一个维护良好的 boto3 变体，称为 aibotocore，您可以找到 here。