오브젝트 스토리지 추상화로 DAG에서 S3·GCS·Azure를 한결같이 쓰기

클라우드 제공자마다 오브젝트 스토리지 접근 방식이 제각각이라, DAG를 짤 때마다 다른 코드를 써야 하는 건 정말 번거로워요. Airflow는 s3, gcs, azure blob storage 같은 다양한 오브젝트 스토리지를 하나의 통일된 인터페이스로 다루는 추상화 계층을 제공해서, 이 문제를 깔끔하게 풀어줘요.

출처: Object Storage — Airflow Documentation

본문

오브젝트 스토리지는 진짜 파일시스템이 아니에요

주요 클라우드 제공자들은 오브젝트 스토어에 데이터를 영구 저장해요. 이 저장소는 우리가 흔히 아는 "POSIX" 파일시스템과는 달라요. 수백 페타바이트의 데이터를 단일 실패 지점 없이 저장하려다 보니, 기존 파일시스템의 디렉터리 트리를 버리고 객체 이름 → 데이터라는 더 단순한 모델을 쓰게 됐어요.

Airflow는 이 오브젝트 스토어들 위에 공통 추상화를 제공해요. 덕분에 DAG 안에서 다양한 오브젝트 스토리지 시스템을, 시스템마다 코드를 바꾸지 않고 그대로 쓸 수 있어요. 어떤 특정 오브젝트 스토리지를 지원하는지는 설치된 프로바이더에 따라 달라져요. 예를 들어 apache-airflow-providers-google 프로바이더를 설치했다면 gcs 스킴을, 설치하지 않았더라도 file 스킴은 기본으로 쓸 수 있어요.

Airflow는 fsspec을 기반으로 여러 오브젝트 스토리지 시스템에서 일관된 경험을 제공하며, 접근 속도를 높이기 위해 로컬 파일 캐시도 구현해요.

기본 사용법

오브젝트 스토리지를 쓰려면 먼저 Path 객체를 다루는 ObjectStoragePath를 만들고, 원하는 객체의 URI를 넣어주면 돼요. s3의 버킷을 가리키려면 이렇게 해요:

from airflow.sdk import ObjectStoragePath

base = ObjectStoragePath("s3://aws_default@my-bucket/")

URI의 사용자 이름 부분은 Airflow 커넥션 id를 의미하며 생략 가능해요. 대신 별도의 키워드 인자로 넘길 수도 있어요:

# 위 예제와 동일한 의미
base = ObjectStoragePath("s3://my-bucket/", conn_id="aws_default")

ObjectStoragePath는 태스크 간에 그대로 주고받으며 쓸 수 있어요. 파일을 만들고 쓰는 태스크를 예로 들어볼게요:

@task
def create(path: ObjectStoragePath) -> ObjectStoragePath:
    return path / "new_file.txt"

@task
def write_file(path: ObjectStoragePath, content: str):
    with path.open("wb") as f:
        f.write(content)

new_file = create(base)
write = write_file(new_file, b"data")

설정

기본 사용에서는 오브젝트 스토리지 추상화가 크게 설정을 요구하지 않고, 표준 Airflow 커넥션 메커니즘에 기대요. conn_id 인자로 쓸 커넥션을 지정하면, 커넥션의 설정이 아래 구현체로 그대로 전달돼요. 예를 들어 s3를 쓸 때 aws_access_key_id, aws_secret_access_key를 지정하고, 추가로 endpoint_url 같은 확장 인자로 커스텀 엔드포인트를 줄 수도 있어요.

대체 백엔드 활용

특정 스킴이나 프로토콜에 대해 대체 백엔드를 붙이는 것도 가능해요. 스킴에 backend를 연결하면 되는데, dbfs 스킴에 databricks 백엔드를 활성화하려면 이렇게 해요:

from airflow.sdk import ObjectStoragePath
from airflow.sdk.io import attach

from fsspec.implementations.dbfs import DBFSFileSystem

attach(protocol="dbfs", fs=DBFSFileSystem(instance="myinstance", token="mytoken"))
base = ObjectStoragePath("dbfs://my-location/")

등록을 여러 태스크에서 재사용하려면, 백엔드를 DAG 최상위 레벨에서 붙여줘야 해요. 안 그러면 백엔드가 여러 태스크에 걸쳐서는 쓰이지 않아요.

Path API

오브젝트 스토리지 추상화는 보통 로컬 파일시스템을 다루듯 쓰는 것을 목표로 해요. Python 표준 Path API와 Universal Pathlib을 기반으로 만들어져서, 로컬 파일시스템과 거의 같은 API로 오브젝트 스토리지를 조작할 수 있어요. 표준 Path API와 다른 점만 따로 정리되어 있으며, 복사나 이동처럼 확장된 연산은 별도 항목으로 다뤄요.

확장 연산

read_block은 파일의 특정 오프셋부터 length 바이트만큼 읽어요. delimiter를 설정하면, 오프셋과 offset + length 위치에 이어지는 구분자 경계에서 읽기가 시작하고 끝나도록 보장해 줘요. 오프셋이 0이면 0부터 시작하고, 반환되는 바이트열에는 끝 구분자 문자열이 포함돼요.

복사와 이동

copy와 move는 source에서 target 위치로 파일이나 디렉터리를 복사·이동하는 연산이에요. 특히 파일 스토리지에서 s3로 넘어가는 등 서로 다른 오브젝트 스토어 간 동작을 다루며, 의도된 동작은 fsspec이 정의한 것과 동일해요.

외부 통합

DuckDB, Apache Iceberg 같은 많은 프로젝트가 오브젝트 스토리지 추상화를 함께 쓸 수 있어요. 보통은 내부 fsspec 구현을 넘겨받는 방식으로 동작하는데, 이를 위해 ObjectStoragePath는 fs 프로퍼티를 노출해요.

오브젝트 스토리지 추상화로 DAG에서 S3·GCS·Azure를 한결같이 쓰기