本教程详细介绍了如何使用 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 中特定前缀（即文件夹）下符合特定命名模式的文件，我们需要：
正确解析 S3 URL，提取桶名和前缀。使用 boto3.resource 获取桶对象。利用 bucket.objects.filter(Prefix=…) 过滤出指定前缀下的所有对象。遍历这些对象，并根据文件名进行模式匹配。
以下是一个核心函数的实现：
立即学习“Python免费学习笔记（深入）”；
import boto3import refrom urllib.parse import urlparsedef 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, prefixdef 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 csvimport boto3import refrom 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_cntFinancial_Freedom_Course_Kannada/00_Course_Trailer_New_update/144p/,28Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/360p/,34Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/480p/,54Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/720p/,57
预期输出 file_count_result.csv 示例：
URL,Actual Files,Expected Files,Statuss3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/144p/,28,28,Successs3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/360p/,34,34,Successs3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/480p/,52,54,Mismatchs3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/720p/,57,57,Success
注意事项
权限管理: 确保您的 AWS 凭证（通过环境变量、IAM 角色或 ~/.aws/credentials 文件配置）具有访问目标 S3 桶和执行 s3:ListBucket 操作的权限。前缀匹配: Prefix 参数是前缀匹配，它会返回所有以该字符串开头的对象。如果您的前缀是 myfolder/，它将匹配 myfolder/file.txt 和 myfolder/subfolder/another.txt。因此，在遍历结果时，仍需通过 obj.key.split(‘/’)[-1] 提取文件名并进行精确匹配。性能优化: 对于包含数百万甚至数十亿对象的超大型桶，即使 boto3.resource 自动处理分页，遍历所有对象仍可能非常耗时。在这种情况下，可以考虑：S3 Inventory: S3 Inventory 服务可以定期生成桶中所有对象的列表，并将其存储在另一个 S3 桶中，这比实时 API 调用更高效。S3 Select: 如果您只需要查询对象内容的特定部分，S3 Select 允许您使用 SQL 表达式直接查询 S3 对象数据。MaxKeys 参数: 在 client 接口中，可以通过 MaxKeys 限制每次 API 调用的返回对象数量，但这需要您手动管理分页。错误处理: 在实际应用中，务必添加健壮的错误处理机制，例如 try-except 块来捕获网络问题、权限不足
以上就是使用 Python 和 Boto3 在 AWS S3 中高效统计指定文件的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1368314.html
微信扫一扫 
支付宝扫一扫 
