如何从AWS S3层级文件夹批量读取JSON文件，检索关键词并返回文件名？

先修复原代码的问题 #

原代码存在几个明显问题：

• 变量名大小写不一致：Keys_list 和 keys_list 混用
• 函数调用错误：self.get_keys_from_prefix 应调用你定义的 fetch_files_from_AWS_S3
• 硬编码密钥不安全：建议用环境变量或AWS默认凭证链替代
• s3_file_read 中的 self 无对应类结构，改成普通函数更合理

完整实现代码 #

下面是整合后的完整代码，包含文件列表获取、文件内容读取、关键词检查三个核心步骤：

import boto3
import json
from typing import List

# 推荐依赖AWS默认凭证链（如EC2角色、~/.aws/credentials、环境变量），避免硬编码密钥
s3_client = boto3.client('s3')

def fetch_files_from_s3(bucket: str, prefix: str) -> List[str]:
    """获取指定S3存储桶和前缀下的所有JSON文件路径"""
    keys_list = []
    # 使用list_objects_v2替代旧API list_objects，性能更优
    paginator = s3_client.get_paginator('list_objects_v2')
    # Delimiter='/' 限定只获取当前层级文件，不递归子文件夹
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix, Delimiter='/'):
        if 'Contents' in page:
            # 过滤出JSON文件，减少无效处理
            json_keys = [content['Key'] for content in page['Contents'] if content['Key'].endswith('.json')]
            print(f"当前页找到 {len(json_keys)} 个JSON文件")
            keys_list.extend(json_keys)
    return keys_list

def check_keyword_in_s3_json(bucket: str, file_key: str, keyword: str) -> bool:
    """检查单个S3 JSON文件中是否包含指定关键词（支持嵌套结构）"""
    try:
        # 小文件直接读取，大文件建议分段处理
        response = s3_client.get_object(Bucket=bucket, Key=file_key)
        content = response['Body'].read().decode('utf-8')
        json_data = json.loads(content)
        
        # 递归遍历JSON所有字段值
        def search_nested(data):
            if isinstance(data, dict):
                for k, v in data.items():
                    if keyword in str(k) or search_nested(v):
                        return True
            elif isinstance(data, list):
                for item in data:
                    if search_nested(item):
                        return True
            else:
                if keyword in str(data):
                    return True
            return False
        
        return search_nested(json_data)
    except Exception as e:
        print(f"处理文件 {file_key} 出错: {str(e)}")
        return False

def find_files_with_keyword(bucket: str, prefix: str, keyword: str) -> List[str]:
    """主函数：返回所有包含指定关键词的JSON文件名"""
    matching_files = []
    file_keys = fetch_files_from_s3(bucket, prefix)
    
    for key in file_keys:
        if check_keyword_in_s3_json(bucket, key, keyword):
            matching_files.append(key)
            print(f"找到匹配文件: {key}")
    
    return matching_files

# 使用示例
if __name__ == "__main__":
    AWS_S3_BUCKET = "你的存储桶名称"
    PREFIX = "s3-bucket-folder-name/"
    TARGET_KEYWORD = "需要查找的关键词"
    
    result = find_files_with_keyword(AWS_S3_BUCKET, PREFIX, TARGET_KEYWORD)
    print("\n所有匹配文件:")
    for file in result:
        print(file)

关键优化说明 #

• API选择：list_objects_v2是官方推荐的列表查询API，性能和功能更完善
• 文件过滤：仅处理.json后缀文件，避免无效操作
• 嵌套检索：递归遍历JSON所有层级，不会遗漏深层字段中的关键词
• 错误隔离：单个文件处理失败不中断整体流程，输出错误信息便于排查
• 凭证安全：摒弃硬编码密钥，依赖AWS默认凭证机制更合规

性能优化建议 #

针对每个子文件夹1000个文件的场景，可进一步优化：

• 用多线程/多进程并行处理文件，提升检索速度
• 超大JSON文件改用分段读取，避免内存溢出
• 用S3 Select直接查询JSON中的关键词，减少数据传输量（适合结构化JSON）

内容的提问来源于stack exchange，提问作者Joe