AWS新手：如何检查S3存储桶文件夹中指定CSV文件是否存在并处理

问题描述 #

作为AWS新手，我需要实现以下逻辑：

1. 检查S3存储桶某个文件夹中是否存在特定CSV文件
2. 若文件存在则读取该文件
3. 若不存在则创建DataFrame并上传至S3

我目前写了部分代码，但判断逻辑有问题：

list_of_files = []
for key in s3_client.list_objects(Bucket= 'abc',Prefix="folder/")['Contents']:
    list_of_files.append(key['Key'])

check_files = [list of file to check]

尝试的判断逻辑：

if set(check_files) in set(list_of_files):
   read_from_s3(file)
else:
  pd.Dataframe()

错误分析 #

你原来的判断逻辑set(check_files) in set(list_of_files)是错误的——这是在判断整个check_files集合是否是list_of_files集合中的单个元素，而不是检查文件是否存在于列表中。

正确实现方案 #

1. 安全获取S3文件列表 #

首先修复文件列表获取逻辑，避免文件夹为空时抛出KeyError（此时list_objects返回的结果中没有Contents字段）：

import boto3
import pandas as pd
from io import BytesIO

# 初始化S3客户端
s3_client = boto3.client('s3')
bucket_name = 'abc'
prefix = 'folder/'

# 获取文件夹下所有文件的Key
list_of_files = []
response = s3_client.list_objects(Bucket=bucket_name, Prefix=prefix)
# 用get方法避免KeyError，文件夹为空时返回空列表
for key in response.get('Contents', []):
    list_of_files.append(key['Key'])

2. 单个文件检查与处理 #

如果只需要检查单个特定文件：

target_file = f"{prefix}target.csv"

if target_file in list_of_files:
    # 读取S3上的CSV文件
    s3_response = s3_client.get_object(Bucket=bucket_name, Key=target_file)
    df = pd.read_csv(s3_response['Body'])
    print(f"成功读取文件：{target_file}")
else:
    # 创建空DataFrame（可自定义列结构）
    df = pd.DataFrame(columns=['列1', '列2', '列3'])
    # 将DataFrame转为CSV字节流
    csv_buffer = BytesIO()
    df.to_csv(csv_buffer, index=False)
    csv_buffer.seek(0)  # 重置流指针到开头
    # 上传至S3
    s3_client.put_object(
        Bucket=bucket_name,
        Key=target_file,
        Body=csv_buffer
    )
    print(f"文件不存在，已创建并上传：{target_file}")

3. 多个文件检查与处理 #

如果需要检查多个文件，可分别处理存在和缺失的文件：

check_files = [f"{prefix}file1.csv", f"{prefix}file2.csv"]

# 拆分存在和缺失的文件
existing_files = [file for file in check_files if file in list_of_files]
missing_files = [file for file in check_files if file not in list_of_files]

# 读取所有存在的文件（示例：合并为一个DataFrame）
if existing_files:
    dfs = []
    for file in existing_files:
        s3_response = s3_client.get_object(Bucket=bucket_name, Key=file)
        dfs.append(pd.read_csv(s3_response['Body']))
    combined_df = pd.concat(dfs, ignore_index=True)
    print(f"已读取{len(existing_files)}个文件")

# 为每个缺失的文件创建并上传空DataFrame
if missing_files:
    for file in missing_files:
        df = pd.DataFrame(columns=['列1', '列2', '列3'])
        csv_buffer = BytesIO()
        df.to_csv(csv_buffer, index=False)
        csv_buffer.seek(0)
        s3_client.put_object(Bucket=bucket_name, Key=file, Body=csv_buffer)
        print(f"已创建并上传缺失文件：{file}")

4. 批量检查所有文件是否存在 #

如果需要确认所有检查文件都存在才执行读取：

if set(check_files).issubset(set(list_of_files)):
    # 所有文件都存在，执行读取逻辑
    dfs = []
    for file in check_files:
        s3_response = s3_client.get_object(Bucket=bucket_name, Key=file)
        dfs.append(pd.read_csv(s3_response['Body']))
    combined_df = pd.concat(dfs, ignore_index=True)
else:
    # 至少有一个文件缺失，创建并上传所有缺失的文件
    missing_files = [file for file in check_files if file not in list_of_files]
    for file in missing_files:
        df = pd.DataFrame(columns=['列1', '列2', '列3'])
        csv_buffer = BytesIO()
        df.to_csv(csv_buffer, index=False)
        csv_buffer.seek(0)
        s3_client.put_object(Bucket=bucket_name, Key=file, Body=csv_buffer)

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