S3 와 S3 호환 스토리지(MinIO · Ceph RGW · Ozone S3 게이트웨이) 를 파이썬에서 다루는 기본 패턴을 정리한다. 목록 조회 · 업로드 · 내려받아 가공 후 다시 올리기, 그리고 Spark 에서 s3a 경로를 다루는 경우를 다룬다.
자격증명은 코드에 적지 않는다. 환경변수 · ~/.aws/credentials · 인스턴스 역할 중 하나를 쓰면 boto3 가 알아서 찾는다.
import boto3
s3 = boto3.client("s3")
S3 호환 스토리지는 엔드포인트를 지정한다. 대부분 경로 스타일 접근이 필요하다.
import boto3
from botocore.client import Config
s3 = boto3.client(
"s3",
endpoint_url="https://minio.example.com:9000",
config=Config(s3={"addressing_style": "path"}),
)
객체가 1000개를 넘으면 한 번의 호출로 다 오지 않는다. 페이지네이터를 쓰면 이어받기를 알아서 처리한다.
paginator = s3.get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket="mybucket", Prefix="data/2026/"):
for obj in page.get("Contents", []):
print(obj["Key"], obj["Size"])
list_objects_v2 를 직접 부르고 Contents 만 읽으면 조용히 1000개에서 잘린다. 흔한 실수다.
"디렉터리" 단위로 보려면 구분자를 준다. S3 에는 디렉터리가 없고 키 이름의 공통 접두사를 그렇게 보여 주는 것뿐이다.
resp = s3.list_objects_v2(Bucket="mybucket", Prefix="data/", Delimiter="/")
for p in resp.get("CommonPrefixes", []):
print(p["Prefix"])
s3.upload_file("/tmp/report.csv", "mybucket", "reports/2026/report.csv")
s3.download_file("mybucket", "reports/2026/report.csv", "/tmp/report.csv")
upload_file 은 큰 파일을 알아서 멀티파트로 나눠 올리고 재시도도 처리한다. put_object 는 본문을 통째로 메모리에 올리므로 작은 객체에만 쓴다.
메모리 위의 데이터를 바로 올릴 때는 upload_fileobj 를 쓴다.
import io
buf = io.BytesIO(b"a,b,c\n1,2,3\n")
s3.upload_fileobj(buf, "mybucket", "tmp/sample.csv")
CSV 의 머리글을 떼고 다른 키로 저장하는 예다. 전체를 메모리에 올리지 않도록 스트리밍으로 처리한다.
import csv, io, json
src = s3.get_object(Bucket="mybucket", Key="in/a.csv")
text = io.TextIOWrapper(src["Body"], encoding="utf-8", newline="")
reader = csv.reader(text)
header = next(reader)
out = io.StringIO()
writer = csv.writer(out)
rows = []
for row in reader:
writer.writerow(row)
rows.append(dict(zip(header, row)))
s3.put_object(Bucket="mybucket", Key="out/b.csv", Body=out.getvalue().encode("utf-8"))
s3.put_object(
Bucket="mybucket",
Key="out/b.json",
Body=json.dumps(rows, ensure_ascii=False).encode("utf-8"),
ContentType="application/json",
)
get_object 의 Body 는 스트리밍 객체다. 한 번만 읽을 수 있으므로 두 번 쓰려면 변수에 담아 둔다.
파일이 크면 이 방식은 맞지 않는다. 로컬로 받아 처리하거나 S3 Select · Athena · Spark 같은 수단을 쓴다.
SparkSession 에 자격증명과 엔드포인트를 설정한다. 값은 코드에 쓰지 말고 환경변수나 설정 파일에서 읽는다.
import os
from pyspark.sql import SparkSession
spark = (
SparkSession.builder.appName("s3-list")
.config("spark.hadoop.fs.s3a.access.key", os.environ["S3_ACCESS_KEY"])
.config("spark.hadoop.fs.s3a.secret.key", os.environ["S3_SECRET_KEY"])
.config("spark.hadoop.fs.s3a.endpoint", "https://minio.example.com:9000")
.config("spark.hadoop.fs.s3a.path.style.access", "true")
.getOrCreate()
)
읽기는 평범하다.
df = spark.read.option("header", True).csv("s3a://mybucket/data/")
목록만 필요하다면 Hadoop FileSystem API 를 py4j 로 부른다. 데이터를 읽지 않으므로 훨씬 빠르다.
jvm = spark._jvm
conf = spark._jsc.hadoopConfiguration()
path = jvm.org.apache.hadoop.fs.Path("s3a://mybucket/data/")
fs = path.getFileSystem(conf)
for st in fs.listStatus(path):
print(st.getPath().toString(), st.getLen())
FileSystem.get(conf) 대신 path.getFileSystem(conf) 를 쓴다. 전자는 기본 파일 시스템(HDFS 등) 을 돌려줄 수 있어 경로와 어긋난다.
s3a 를 쓰려면 hadoop-aws 와 그에 맞는 aws-java-sdk jar 가 클래스패스에 있어야 하고, 두 버전이 정확히 맞아야 한다. 어긋나면 NoSuchMethodError 계열로 실패한다.