如何批量读取S3 bucket中的JSON对象并加载至pandas DataFrame

前置准备 #

首先安装需要的Python包，同时确保运行环境已经配置好目标S3桶的读取权限：

• 本地运行可以通过aws configure配置访问密钥，或者设置AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY环境变量
• AWS云上环境（EC2、Lambda、SageMaker等）直接给运行角色绑定S3读取权限即可，无需手动配置密钥
• 安装依赖命令：pip install boto3 pandas s3fs

方案1：Boto3原生实现（可控性高，适配大文件量场景） #

这个方案通过S3 API分页遍历对象，支持自定义过滤规则，不会因为桶内对象数超过1000个出现漏读，适合生产环境使用。

import boto3
import pandas as pd
import json
from io import BytesIO

# 初始化S3客户端，注意替换为你桶所在的实际区域
s3_client = boto3.client("s3", region_name="ap-east-1")
BUCKET_NAME = "替换成你的桶名"
# 如果JSON文件都存放在指定目录下，填写前缀即可，比如"raw_data/2024/"，没有层级就留空
FILE_PREFIX = ""

record_list = []

# 启用分页遍历，解决S3单次接口最多返回1000个对象的限制
paginator = s3_client.get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket=BUCKET_NAME, Prefix=FILE_PREFIX):
    if "Contents" not in page:
        continue
    for obj in page["Contents"]:
        file_key = obj["Key"]
        # 过滤掉非JSON文件、S3自动生成的文件夹占位对象
        if file_key.endswith("/") or not file_key.endswith(".json"):
            continue
        # 读取对象内容
        file_obj = s3_client.get_object(Bucket=BUCKET_NAME, Key=file_key)
        file_content = file_obj["Body"].read()
        # 解析JSON，根据你的文件实际结构调整逻辑
        json_content = json.load(BytesIO(file_content))
        # 如果单个JSON文件存的是多条记录的数组，用extend；单条记录用append
        if isinstance(json_content, list):
            record_list.extend(json_content)
        else:
            record_list.append(json_content)

# 转换为DataFrame
final_df = pd.DataFrame(record_list)

适配说明 #

• 如果你存储的是JSON Lines格式（每行一条JSON对象，大数据场景常用），把解析部分替换为temp_df = pd.read_json(BytesIO(file_content), lines=True)，再把小df追加到列表里，最后统一concat，解析效率更高。
• 可以在遍历过程中加自定义过滤逻辑，比如只读取某个日期之后上传的文件、跳过特定前缀的文件等。

方案2：S3FS+Pandas简写（代码简洁，适配小批量文件场景） #

如果你的JSON文件结构统一，文件量不大，可以直接用s3fs做文件系统适配，用pandas原生接口读取，代码量更少：

import pandas as pd
import s3fs

# 初始化S3文件系统，自动读取本地配置的凭证
fs = s3fs.S3FileSystem(anon=False)
BUCKET_NAME = "替换成你的桶名"
FILE_PREFIX = ""

# 匹配前缀下所有JSON文件
file_list = fs.glob(f"s3://{BUCKET_NAME}/{FILE_PREFIX}*.json")

df_collector = []
for file_path in file_list:
    # 普通JSON去掉lines参数，JSON Lines格式保留lines=True
    temp_df = pd.read_json(f"s3://{file_path}", lines=True)
    df_collector.append(temp_df)

final_df = pd.concat(df_collector, ignore_index=True)

注意事项 #

• 这个方案会一次性拉取全量匹配的文件列表，桶内文件数过万时内存开销更高，优先选方案1。
• 运行报错优先检查权限：需要配置的权限包含s3:ListBucket（列文件）和s3:GetObject（读文件），缺任意一个都会失败。
• 如果不同JSON文件的字段不一致，拼接时Pandas会自动给缺失字段填充NaN，需要做字段清洗可以在读取单文件时提前处理。

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