Airflow

객체 스토어는 진짜 파일시스템이 아니다

대부분의 주요 클라우드 프로바이더는 영구 데이터 저장소로 객체 스토어를 제공한다. 이는 고전적인 'POSIX' 파일시스템이 아니다. 단일 장애 지점 없이 수백 페타바이트를 저장하기 위해, 객체 스토어는 고전적인 디렉터리 트리 파일시스템 대신 객체 이름 => 데이터라는 더 단순한 모델을 쓴다. 원격 접근을 위해 객체에 대한 연산은 보통 (느린) HTTP REST 연산으로 제공된다.

Airflow는 s3, gcs, azure blob storage 같은 다양한 객체 스토리지 시스템 위에 범용 추상화를 제공한다. 이를 통해 Dag 코드를 바꾸지 않고도 다양한 객체 스토리지 시스템을 쓸 수 있고, shutil처럼 파일류 객체를 다루는 표준 Python 모듈 대부분도 함께 쓸 수 있다. 특정 객체 스토리지 시스템 지원 여부는 설치한 provider에 달려 있다 — 기본적으로는 file 스킴을 지원한다.

객체 스토어는 겉보기와 달리 진짜 파일시스템이 지원하는 모든 연산을 지원하지 않는다. 핵심 차이는 다음과 같다. 원자적인 rename 연산이 보장되지 않는다 — 파일을 한 위치에서 다른 위치로 옮기면 실제로는 복사 후 삭제되며, 복사가 실패하면 파일을 잃는다. 디렉터리는 에뮬레이션된 것이라 다루기 느릴 수 있다 — 예를 들어 디렉터리 목록을 나열하려면 버킷의 모든 객체를 나열한 뒤 prefix로 필터링해야 할 수 있다. 파일 내부에서 seek하는 것은 호출 오버헤드가 커서 성능을 해치거나, 아예 지원되지 않을 수도 있다.

Airflow는 다양한 객체 스토리지 시스템에서 일관된 경험을 제공하기 위해 fsspec에 의존하며, 접근 속도를 높이기 위해 로컬 파일 캐싱을 구현한다. 그럼에도 Dag를 설계할 때는 객체 스토리지의 이러한 한계를 인지하고 있어야 한다.

기본 사용법과 백엔드 설정

객체 스토리지를 쓰려면 상호작용할 객체의 URI로 Path 객체를 만들면 된다. 예를 들어 s3 버킷을 가리키려면 `ObjectStoragePath("s3://aws_default@my-bucket/")`처럼 쓴다 — URI의 사용자명 부분은 Airflow 커넥션 ID를 의미하며 선택적이고, 대신 `conn_id` 키워드 인자로 별도로 넘길 수도 있다. 지원 여부는 설치한 provider에 달려 있다 — 예를 들어 apache-airflow-providers-google을 설치하면 gcs 스킴을 쓸 수 있다. file 스킴은 기본 제공된다. s3 지원에는 apache-airflow-providers-amazon[s3fs] 설치가 별도로 필요한데, 이는 aiobotocore에 의존하기 때문이며 botocore와의 의존성 충돌을 피하기 위해 기본으로는 설치되지 않는다.

기본 연산으로는 base.iterdir()로 파일 나열, `/` 연산자로 하위 디렉터리 탐색, .open()으로 파일 열기, 그리고 XCom을 통해 태스크 사이에 경로를 전달하는 것 등이 있다.

대체 백엔드를 프로토콜/스킴에 붙일 수도 있다. 예를 들어 dbfs 스킴에 databricks 백엔드를 연결하려면 `airflow.sdk.io.attach(protocol="dbfs", fs=DBFSFileSystem(...))`를 쓴다. 여러 태스크에서 이 등록을 재사용하려면 반드시 Dag의 최상위 레벨에서 attach해야 한다 — 그렇지 않으면 다른 태스크에서는 이 백엔드를 쓸 수 없다.

Path API와 확장 연산

이 추상화는 Python 표준 Path API(pathlib)와 Universal Pathlib 위에 구축되어, 로컬 파일시스템을 다루던 것과 거의 같은 API로 객체 스토리지를 다룰 수 있다. mkdir은 지정된 경로 또는 버킷/컨테이너 안에 디렉터리 엔트리를 만든다 — 진짜 디렉터리 개념이 없는 시스템에서는 이 인스턴스만을 위한 디렉터리 엔트리를 만들 뿐 실제 파일시스템에는 영향을 주지 않을 수 있다. parents=True면 누락된 상위 경로도 함께 생성한다. touch는 파일을 생성하거나 타임스탬프를 갱신하며, truncate=True(기본값)면 파일을 잘라내고, exists_ok가 true면 이미 존재하는 파일에 대해서도 성공(수정 시각만 갱신)하지만 그렇지 않으면 FileExistsError를 낸다. stat은 st_size, st_mtime, st_mode를 지원하는 stat_result 유사 객체를 반환하면서 동시에 딕셔너리처럼 추가 메타데이터도 제공한다 — 예를 들어 s3에서는 ETag, ContentType 같은 확장 키가 추가되는데, 스토리지 간 이식성이 필요한 코드라면 이런 확장 메타데이터에 의존하지 말아야 한다.

표준 Path API에는 없지만 이 추상화가 추가로 지원하는 연산들도 있다. bucket은 버킷 이름을 반환한다. checksum은 파일의 체크섬을 반환한다. container는 bucket의 별칭이다. fs는 인스턴스화된 파일시스템에 접근하는 편의 속성이다. key는 객체 키를 반환한다. namespace는 객체의 네임스페이스(전형적으로 s3:// 같은 프로토콜과 버킷명)를 반환한다. path는 파일시스템 인스턴스에서 쓸 fsspec 호환 경로다. protocol은 filesystem_spec 프로토콜이다. read_block(offset, length, delimiter)은 파일의 offset부터 length 바이트를 읽되, delimiter가 설정되어 있으면 offset과 offset+length를 뒤따르는 delimiter 경계에서 읽기 시작·종료를 맞춘다(offset이 0이면 처음부터 시작하며, 반환되는 바이트열에는 종료 delimiter 문자열까지 포함된다. offset+length가 파일 끝을 넘으면 파일 끝까지 읽는다). sign은 임시 URL 생성을 지원하는 일부 구현에서, 자격 증명을 위임하는 방식으로 서명된 URL을 만든다. size는 파일 크기를 바이트 단위로 반환한다. storage_options는 하부 파일시스템을 인스턴스화하는 데 쓰인 저장소 옵션이다. ukey는 파일이 변경되었는지 알려주는 파일 속성 해시다.

복사/이동과 외부 통합

copy와 move 연산의 기대 동작은, 특히 서로 다른 객체 스토어 사이(예: file -> s3)에서 fsspec이 규정하는 것과 동일한 방식을 따른다. 서로 다른 스토어 간 디렉터리 복사에서는 Airflow가 디렉터리 트리를 순회하며 각 파일을 소스에서 타깃으로 스트리밍해 개별적으로 복사해야 한다.

DuckDB, Apache Iceberg 같은 다른 여러 프로젝트도 이 객체 스토리지 추상화를 활용할 수 있다. 보통은 하부 fsspec 구현체를 그대로 넘기는 방식으로 이루어지며, 이를 위해 ObjectStoragePath는 fs 속성을 노출한다. 예를 들어 DuckDB가 parquet 파일을 읽을 때, Airflow 커넥션에서 유도된 s3 자격 증명으로 접속하도록 `conn.register_filesystem(path.fs)`를 호출한 뒤 `read_parquet('{path}')`로 조회할 수 있다.