如何在YAML文件中配置S3存储桶内目录路径至source_directory变量

你之前直接用s3://<bucket_name>/object_path报错，核心原因是原来的代码逻辑是把配置值当作本地文件路径处理，但s3://是对象存储的协议路径，普通本地文件操作函数（比如os模块、open()）根本不识别这个格式，必须结合boto3调整代码逻辑，同时可以优化YAML的配置方式。

下面提供两种可行方案：

方案1：YAML拆分存储桶和路径前缀 #

这种方式更清晰，便于后续扩展配置。

修改YAML配置 #

source_s3:
  bucket: "你的存储桶名称"
  prefix: "桶内的目标目录路径/"  # 比如"raw-data/2024/"，结尾加/明确是前缀

调整Python代码 #

import yaml
import boto3

# 加载YAML配置
with open('你的配置文件.yaml') as file:
    config = yaml.load(file, Loader=yaml.FullLoader)

# 初始化S3客户端（已确认连接正常，可直接复用你的客户端初始化逻辑）
s3_client = boto3.client('s3')

# 读取S3配置
s3_config = config['source_s3']
bucket_name = s3_config['bucket']
prefix = s3_config['prefix']

# 列出S3前缀下的所有文件（用分页器处理大量文件场景）
def list_s3_files(client, bucket, prefix):
    paginator = client.get_paginator('list_objects_v2')
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for content in page.get('Contents', []):
            # 跳过前缀本身（如果prefix以/结尾，可能会返回空的目录Key）
            if content['Key'] != prefix:
                yield content['Key']

# 遍历输出文件
file_list = list_s3_files(s3_client, bucket_name, prefix)
for file in file_list:
    print(file)

方案2：YAML保留s3://格式路径 #

如果想继续用source_directory字段存完整的S3路径，需要在代码中解析路径格式。

修改YAML配置 #

source_directory: "s3://你的存储桶名称/桶内的目标目录路径/"

调整Python代码 #

import yaml
import boto3
from urllib.parse import urlparse

# 加载YAML配置
with open('你的配置文件.yaml') as file:
    config = yaml.load(file, Loader=yaml.FullLoader)

source_dir = config['source_directory']

# 解析S3路径
parsed_url = urlparse(source_dir)
if parsed_url.scheme == 's3':
    bucket_name = parsed_url.netloc
    prefix = parsed_url.path.lstrip('/')  # 去掉路径开头的/，适配S3前缀格式

    # 初始化S3客户端
    s3_client = boto3.client('s3')

    # 复用上面的文件列出函数
    def list_s3_files(client, bucket, prefix):
        paginator = client.get_paginator('list_objects_v2')
        for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
            for content in page.get('Contents', []):
                if content['Key'] != prefix:
                    yield content['Key']

    # 遍历输出文件
    file_list = list_s3_files(s3_client, bucket_name, prefix)
    for file in file_list:
        print(file)
else:
    # 保留原本地目录的处理逻辑（兼容旧配置）
    import os
    for root, _, files in os.walk(source_dir):
        for file in files:
            print(os.path.join(root, file))

关键说明 #

• 不要试图用本地文件操作函数直接处理s3://路径，必须通过boto3的API来和S3交互。
• 推荐用list_objects_v2替代list_objects，它是S3的最新API，支持分页和更多功能，避免大量文件时的性能问题。

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