使用 Python 和 Boto3 在 AWS S3 中高效统计指定文件

· 日常编程 · 237510

本教程详细介绍了如何使用 Python 和 Boto3 库高效地统计 AWS S3 存储桶中特定路径下符合命名模式的文件。文章重点阐述了 boto3.resource 相较于 boto3.client 在处理大量对象时的优势（例如自动分页），并提供了从 S3 URL 中提取桶名和前缀的方法。通过结合正则表达式，您可以精确筛选并计数如 file_000.ts 这样的增量文件，并给出了一个完整的示例，演示如何批量处理 S3 路径并输出统计结果。

在 aws s3 中管理大量文件时，经常需要统计特定目录（前缀）下符合某种命名规则的文件数量。例如，视频处理场景中，一个视频文件可能被切分为多个增量块，如 file_000.ts, file_001.ts 等，并存储在不同质量（如 144p, 360p）的嵌套文件夹中。本教程将指导您如何使用 python 和 boto3 库来自动化这一过程。

Boto3 客户端与资源对象：选择与优势

boto3 提供了两种主要接口来与 AWS 服务交互：client 和 resource。

  • client 接口 (低级)：提供与 AWS API 一对一的映射，功能最全面，但通常需要手动处理分页（例如，当 list_objects_v2 返回超过 1000 个对象时）。
  • resource 接口 (高级)：提供更高级、更面向对象的方法，封装了许多底层细节，包括自动处理分页。对于大多数日常操作，尤其是需要遍历大量 S3 对象的场景，resource 接口通常更简洁、更易用。

在统计 S3 文件数量时，由于可能存在大量文件，boto3.resource 自动处理分页的能力使其成为更优的选择，避免了手动检查 IsTruncated 和 NextContinuationToken 的复杂性。

核心实现：统计 S3 中特定前缀下的文件

要准确统计 S3 中特定前缀（即文件夹）下符合特定命名模式的文件，我们需要：

  1. 正确解析 S3 URL，提取桶名和前缀。
  2. 使用 boto3.resource 获取桶对象。
  3. 利用 bucket.objects.filter(Prefix=...) 过滤出指定前缀下的所有对象。
  4. 遍历这些对象，并根据文件名进行模式匹配。

以下是一个核心函数的实现：

import boto3
import re
from urllib.parse import urlparse

def get_bucket_and_prefix_from_s3_url(s3_url: str) -> tuple[str, str]:
    """
    解析 S3 URL 以提取桶名和对象前缀（路径）。

    Args:
        s3_url (str): 完整的 S3 URL，例如 's3://your-bucket-name/path/to/folder/'

    Returns:
        tuple[str, str]: 包含桶名和前缀的元组。

    Raises:
        ValueError: 如果 S3 URL 格式不正确。
    """
    parsed_url = urlparse(s3_url)
    if parsed_url.scheme != 's3':
        raise ValueError("无效的 S3 URL 方案。URL 必须以 's3://' 开头。")

    bucket_name = parsed_url.netloc
    # path.lstrip('/') 移除路径开头的斜杠
    prefix = parsed_url.path.lstrip('/')

    # 确保前缀以斜杠结尾，如果它表示一个文件夹
    if prefix and not prefix.endswith('/'):
        prefix += '/'

    return bucket_name, prefix

def count_specific_files_in_s3(
    bucket_name: str, 
    s3_prefix: str, 
    file_regex_pattern: str = r'^file_\d+\.ts$'
) -> int:
    """
    统计 S3 存储桶中指定前缀下符合正则表达式模式的文件数量。

    Args:
        bucket_name (str): S3 存储桶的名称。
        s3_prefix (str): S3 前缀（文件夹路径），例如 'path/to/folder/'。
        file_regex_pattern (str): 用于匹配对象键中“文件名”部分的正则表达式模式。
                                  默认匹配 'file_NNN.ts' 格式的文件。

    Returns:
        int: 匹配文件的数量。如果发生错误，返回 0。
    """
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(bucket_name)

    count = 0
    # 编译正则表达式以提高性能
    file_pattern = re.compile(file_regex_pattern)

    try:
        # 使用 filter 方法和 S3 前缀来获取指定路径下的所有对象
        # boto3.resource 会自动处理分页
        for obj in bucket.objects.filter(Prefix=s3_prefix):
            # 提取对象键（完整路径）中的文件名部分
            filename = obj.key.split('/')[-1]

            # 检查文件名是否与期望的模式匹配
            if file_pattern.match(filename):
                count += 1
    except Exception as e:
        print(f"错误：在 s3://{bucket_name}/{s3_prefix} 统计文件时发生异常: {e}")
        return 0 # 发生错误时返回 0，或根据需要抛出异常

    return count

在上述代码中：

  • get_bucket_and_prefix_from_s3_url 函数负责将一个完整的 S3 URL 分解为桶名和在桶内的对象前缀。这是关键一步，因为 boto3 的 Prefix 参数期望的是桶内的路径，而不是完整的 S3 URL。
  • count_specific_files_in_s3 函数接收桶名和前缀，并使用 boto3.resource('s3').Bucket(bucket_name).objects.filter(Prefix=s3_prefix) 来高效地遍历所有匹配的对象。
  • 我们使用 re.compile 预编译正则表达式，并通过 file_pattern.match(filename) 来精确匹配文件名，确保只统计符合 file_NNN.ts 模式的文件。

完整示例：结合 CSV 处理批量统计

在实际应用中，您可能需要从一个 CSV 文件中读取多条 S3 路径，然后对每条路径进行文件统计，并将结果写入另一个 CSV 文件。以下是一个结合上述函数的完整示例：

import csv
import boto3
import re
from urllib.parse import urlparse

# (此处省略上面定义的 get_bucket_and_prefix_from_s3_url 和 count_specific_files_in_s3 函数)
# 请确保将上述两个函数定义复制到此处或导入

# 定义 S3 URL 的基础部分，如果您的 CSV 中只包含相对路径
base_s3_url = 's3://coursevideotesting/' 

# 输入和输出 CSV 文件名
input_csv_file = 'ldt_ffw_course_videos_temp.csv'
output_csv_file = 'file_count_result.csv'

# 定义要匹配的文件正则表达式模式
# 例如，匹配 'file_000.ts', 'file_001.ts' 等
TARGET_FILE_PATTERN = r'^file_\d+\.ts$' 

# 读取输入 CSV 并检查文件数量
with open(input_csv_file, mode='r', encoding='utf-8') as infile, \
     open(output_csv_file, mode='w', newline='', encoding='utf-8') as outfile:

    reader = csv.DictReader(infile)
    # 确保 CSV 字段名与您的输入文件匹配
    # 例如，如果您的输入 CSV 有 'course_video_s3_url' 和 'course_video_ts_file_cnt' 列
    fieldnames = ['URL', 'Actual Files', 'Expected Files', 'Status'] # 添加一个状态列
    writer = csv.DictWriter(outfile, fieldnames=fieldnames)
    writer.writeheader()

    for row in reader:
        # 构建完整的 S3 URL
        # 假设 input_csv_file 中的 'course_video_s3_url' 包含相对路径，如 'Financial_Freedom_Course_Kannada/...'
        full_s3_url = base_s3_url + row['course_video_s3_url']

        expected_files = int(row['course_video_ts_file_cnt'])
        actual_files = 0
        status = "Success"

        try:
            # 解析 S3 URL 获取桶名和前缀
            bucket_name, s3_prefix = get_bucket_and_prefix_from_s3_url(full_s3_url)

            # 调用函数统计文件
            actual_files = count_specific_files_in_s3(bucket_name, s3_prefix, TARGET_FILE_PATTERN)

            if actual_files != expected_files:
                status = "Mismatch"
            if actual_files == 0 and expected_files > 0:
                status = "Missing" # 预期有文件但实际没有

        except ValueError as ve:
            print(f"处理 URL 格式错误: {full_s3_url} - {ve}")
            status = f"URL Error: {ve}"
        except Exception as e:
            print(f"处理 URL 时发生未知错误: {full_s3_url} - {e}")
            status = f"Process Error: {e}"

        writer.writerow({
            'URL': full_s3_url, 
            'Actual Files': actual_files, 
            'Expected Files': expected_files,
            'Status': status
        })

print(f"文件统计结果已写入到 {output_csv_file}")

输入 CSV 文件 ldt_ffw_course_videos_temp.csv 示例：

course_video_s3_url,course_video_ts_file_cnt
Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/144p/,28
Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/360p/,34
Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/480p/,54
Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/720p/,57

预期输出 file_count_result.csv 示例：

URL,Actual Files,Expected Files,Status
s3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/144p/,28,28,Success
s3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/360p/,34,34,Success
s3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/480p/,52,54,Mismatch
s3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/720p/,57,57,Success

注意事项

  1. 权限管理: 确保您的 AWS 凭证（通过环境变量、IAM 角色或 ~/.aws/credentials 文件配置）具有访问目标 S3 桶和执行 s3:ListBucket 操作的权限。
  2. 前缀匹配: Prefix 参数是前缀匹配，它会返回所有以该字符串开头的对象。如果您的前缀是 myfolder/，它将匹配 myfolder/file.txt 和 myfolder/subfolder/another.txt。因此，在遍历结果时，仍需通过 obj.key.split('/')[-1] 提取文件名并进行精确匹配。
  3. 性能优化: 对于包含数百万甚至数十亿对象的超大型桶，即使 boto3.resource 自动处理分页，遍历所有对象仍可能非常耗时。在这种情况下，可以考虑：

    • S3 Inventory: S3 Inventory 服务可以定期生成桶中所有对象的列表，并将其存储在另一个 S3 桶中，这比实时 API 调用更高效。
    • S3 Select: 如果您只需要查询对象内容的特定部分，S3 Select 允许您使用 SQL 表达式直接查询 S3 对象数据。
    • MaxKeys 参数: 在 client 接口中，可以通过 MaxKeys 限制每次 API 调用的返回对象数量，但这需要您手动管理分页。
  4. 错误处理: 在实际应用中，务必添加健壮的错误处理机制，例如 try-except 块来捕获网络问题、权限不足

以上就是使用 Python 和 Boto3 在 AWS S3 中高效统计指定文件的详细内容，更多请关注骃骐网【www.myinqi.com】。

相关推荐：

Python如何高效搜索大型日志文件中的特定错误_利用mmap内存映射加速查找

直接用 open() 逐行读取大日志会卡住，因需全量解码、切分和构造字符串；mmap 通过内存映射实现字节级随机访问，配合 bytes 搜索或 re.DOTALL 正则可高效定位关键词，避免冗余解析与跨块漏匹配。 为什么直接用 open() 逐行读取大日志会卡住 10GB 以上的日志文件，用 for line in open(...) 或 readlines() 会触发大量磁盘 I/O 和内存分配...

如何高效流式读取大型CSV文件并异步发送HTTP请求

本文介绍针对20GB级CSV文件的内存友好型流式处理方案，通过csv.DictReader逐行解析、aiohttp并发控制与XML模板填充，实现低内存占用、高吞吐量的HTTP批量提交。 本文介绍针对20gb级csv文件的内存友好型流式处理方案，通过`csv.dictreader`逐行解析、`aiohttp`并发控制与xml模板填充，实现低内存占用、高吞吐量的http批量提交。 处理超大CSV文件（...

高效实现 Pandas DataFrame 中基于多列的条件列创建

本文介绍多种优化方法，帮助你在处理大规模数据时高效创建条件列，包括向量化操作、numpy.select、pd.cut 和缓存函数调用等策略，显著减少计算开销并提升可读性与执行速度。 本文介绍多种优化方法，帮助你在处理大规模数据时高效创建条件列，包括向量化操作、`numpy.select`、`pd.cut` 和缓存函数调用等策略，显著减少计算开销并提升可读性与执行速度。 在 Pandas 中为大型 ...

如何在Python中实现高效的相似度搜索_使用Faiss向量数据库加速

Faiss将向量搜索从O(N)降至近O(log N)，通过索引、量化与压缩实现加速；选错索引类型会导致构建失败或结果不准，小规模用IndexFlatL2，中大规模依数据量选IVF或HNSW。 为什么直接用 NumPy 算余弦相似度会变慢 当向量数量超过 10 万，用 sklearn.metrics.pairwise.cosine_similarity 或手写 np.dot + 归一化做全量比对，内...

如何高效实现可配置窗口大小的移动平均滤波器

本文介绍如何用 numpy 和 pandas 实现灵活、高效的移动平均函数，支持任意窗口大小配置，避免硬编码索引，显著提升性能与可维护性。 本文介绍如何用 numpy 和 pandas 实现灵活、高效的移动平均函数，支持任意窗口大小配置，避免硬编码索引，显著提升性能与可维护性。 在信号处理、时间序列分析和数据平滑任务中，移动平均是一种基础而关键的滤波技术。原始实现中将窗口大小（如 12）硬编码在索...

如何高效检查字符串中每个数字出现次数是否均不超过4次

本文介绍如何用一行python代码高效验证字符串中所有数字的出现频次是否均≤4，避免重复遍历字符串，提升性能。 本文介绍如何用一行python代码高效验证字符串中所有数字的出现频次是否均≤4，避免重复遍历字符串，提升性能。 在处理字符串校验任务时，若需确保每个数字（0–9）在字符串中最多出现4次，直接对每个数字调用 s.count('d') 是常见但低效的做法——它会导致字符串被遍历多达10次（每...

如何高效提取 DataFrame 中指定起始索引的连续行段（支持单列与多列）

本文介绍一种高效、向量化的方法，替代低效的循环 pd.concat，从原始 DataFrame 中按子集索引批量提取长度为 n 的连续行段，尤其适用于单列选取场景，避免因误用 .iloc 与 [] 操作符导致的索引对齐错误和 NaN 问题。 本文介绍一种高效、向量化的方法，替代低效的循环 `pd.concat`，从原始 dataframe 中按子集索引批量提取长度为 `n` 的连续行段，尤其适用于...

如何在Python Django中实现分库分表逻辑_利用Router控制模型路由到指定库

Django的DatabaseRouter仅支持分库，无法自动分表，因其只在查询前选择数据库别名，不干预SQL生成与表名替换，分表需手动处理建表、查询、写入及迁移。 分库分表不能靠 Django 的 DatabaseRouter 完全实现，它只负责“路由到库”，不处理“表拆分逻辑”；真正要分表（比如按时间/用户ID哈希），必须手动干预查询、写入和迁移流程。 为什么 DatabaseRouter 只...

矩阵逐片乘法（Matrix-wise Multiplication）的高效实现

使用 NumPy 的 np.matmul 可对批量矩阵（M×3×3）与对应向量（M×3×1）执行高效逐片矩阵乘法，避免显式 Python 循环，显著提升计算性能。 使用 numpy 的 `np.matmul` 可对批量矩阵（m×3×3）与对应向量（m×3×1）执行高效逐片矩阵乘法，避免显式 python 循环，显著提升计算性能。 在深度学习、几何变换或批量线性系统求解等场景中，常需对一批形状为 M...

Pandas 中按列分组统计满足条件的唯一值数量并筛选结果

本文介绍如何在 Pandas 中基于多条件过滤（如 colC > 0 且 colB 属于指定集合），按 colA 分组统计 colB 的唯一值个数，并仅保留计数大于 2 的分组，最终导出对应原始数据子集。 本文介绍如何在 pandas 中基于多条件过滤（如 `colc > 0` 且 `colb` 属于指定集合），按 `cola` 分组统计 `colb` 的唯一值个数，并仅保留计数大于 ...