如何使用Python将多条print输出重定向为单个JSON格式文件
实现方法
S3 对象本身不支持原地追加操作,你可以按照「拉取已有文件内容 -> 追加新条目 -> 全量写回」的流程实现,具体步骤和可运行代码如下:
- 提前配置好boto3的访问凭证,可通过本地AWS配置文件、EC2/ECS绑定的IAM角色完成授权,不要在代码里硬编码AK/SK
- 读取目标S3路径下已有的JSON文件:如果文件是首次创建(不存在),初始化空列表作为存储容器;如果文件已存在,将内容反序列化为Python列表
- 将你之前提取到的所有包含
author、timestamp字段的元数据条目追加到列表中 - 将整合后的列表序列化为标准JSON格式,调用S3的
put_object接口写回目标路径,写入时指定ContentType为application/json避免后续读取时识别为二进制文件
import boto3 import json from botocore.exceptions import ClientError # 初始化S3客户端 s3 = boto3.client('s3') # 配置项 TARGET_BUCKET = '你的存储桶名' TARGET_JSON_KEY = 'metadata/collected.json' # JSON文件在S3中的存储路径 # 这里替换成你之前读取到的多条元数据,格式为包含author、timestamp字段的字典列表 new_metadata_items = [ {"author": "张三", "timestamp": "2024-05-01T12:00:00Z"}, {"author": "李四", "timestamp": "2024-05-02T13:30:00Z"} ] # 拉取S3上已有的JSON内容 existing_records = [] try: response = s3.get_object(Bucket=TARGET_BUCKET, Key=TARGET_JSON_KEY) existing_content = response['Body'].read().decode('utf-8') existing_records = json.loads(existing_content) # 校验已有内容格式是不是列表,避免格式错误 if not isinstance(existing_records, list): existing_records = [existing_records] except ClientError as e: # 只有文件不存在的错误才忽略,其他错误正常抛出 if e.response['Error']['Code'] != 'NoSuchKey': raise e # 追加新的元数据条目 existing_records.extend(new_metadata_items) # 序列化为JSON字符串,indent=2保证文件可读性 final_json_content = json.dumps(existing_records, ensure_ascii=False, indent=2) # 写回S3 s3.put_object( Bucket=TARGET_BUCKET, Key=TARGET_JSON_KEY, Body=final_json_content.encode('utf-8'), ContentType='application/json' )
注意:如果你的元数据累计量级超过100MB,这种全量拉取再写回的方式性能会明显下降,这种场景建议按日期/批次拆分存储多个JSON文件,不要强行维护单个超大JSON文件。
内容的提问来源于stack exchange,提问作者AMIT SINGH
相关产品推荐
相关产品推荐

