Airflow

왜 Object Storage API인가

Object Storage API는 Airflow 2.8에서 추가되었다. 많은 데이터 워크플로우는 원본 CSV, 중간 산출물인 Parquet 파일, 모델 아티팩트처럼 파일에 의존한다. 전통적으로는 이런 작업을 위해 S3 전용 또는 GCS 전용 코드를 따로 작성해야 했다. 이제 ``ObjectStoragePath``를 쓰면, 올바른 Airflow 커넥션만 구성되어 있다면 여러 프로바이더에 걸쳐 동작하는 범용 코드를 작성할 수 있다 — 즉 클라우드에 종속되지 않는(cloud-agnostic) 파이프라인을 만들 수 있다는 뜻이다.

ObjectStoragePath: 버킷을 위한 pathlib

이 튜토리얼의 핵심은 클라우드 오브젝트 스토어의 경로를 다루기 위한 새로운 추상화인 ``ObjectStoragePath``다. 파일시스템이 아니라 버킷을 위한 ``pathlib.Path``라고 생각하면 된다.

URL 문법은 간단하다: ``protocol://bucket/path/to/file``. ``protocol``(``s3``, ``gs``, ``azure`` 등)이 어떤 백엔드를 쓸지 결정하고, URL의 "사용자 이름" 부분에는 ``conn_id``를 넣어 Airflow에게 어떻게 인증할지 알려줄 수 있다. 만약 ``conn_id``를 생략하면, Airflow는 그 백엔드에 대한 기본 커넥션으로 폴백한다.

``conn_id``를 키워드 인자로 명시적으로 전달할 수도 있다.

ObjectStoragePath("s3://airflow-tutorial-data/", conn_id="aws_default")

이 방식은 다른 곳(예: Asset)에서 정의된 경로를 재사용하거나, 커넥션 정보가 URL 안에 담겨 있지 않을 때 특히 유용하다. 키워드 인자는 URL 안의 값보다 항상 우선한다. ``ObjectStoragePath``는 Dag의 전역 스코프에 안전하게 만들어 둘 수 있다는 점도 알아둘 만하다 — 커넥션은 그 경로가 생성될 때가 아니라 실제로 사용될 때 비로소 해석(resolve)되기 때문이다.

데이터를 클라우드에 저장하기

핀란드 기상연구소의 헬싱키 대기질 공개 API를 호출해 데이터를 가져오고, 그 JSON 응답을 pandas DataFrame으로 파싱한 뒤, 태스크의 logical date를 기반으로 파일명을 생성하고, ``ObjectStoragePath``를 이용해 그 데이터를 Parquet 형식으로 클라우드에 직접 쓴다. 오브젝트 키가 매일 달라지므로, 이 태스크를 매일 실행하면 시간이 지남에 따라 하나의 데이터셋이 누적된다. 최종 오브젝트 경로는 다운스트림 태스크에서 쓸 수 있도록 반환된다.

이 방식이 좋은 이유는, boto3도 없고 GCS 클라이언트 설정도 없고 자격증명을 일일이 다루지 않아도 되며, 여러 스토리지 백엔드에서 동일하게 동작하는 단순한 파일 시맨틱만으로 충분하다는 점이다.

DuckDB로 저장된 데이터 분석하기

DuckDB는 Parquet을 네이티브로 읽을 수 있는 인프로세스(in-process) SQL 데이터베이스다. DuckDB와 ObjectStoragePath는 둘 다 ``fsspec``에 의존하기 때문에, 오브젝트 스토리지 백엔드를 DuckDB에 등록하기가 쉽다 — ``path.fs``로 올바른 파일시스템 객체를 가져와 DuckDB에 등록해주면 된다. 마지막으로 SQL로 Parquet 파일을 쿼리하고 pandas DataFrame을 반환한다.

이때 이 함수는 경로를 직접 재구성하지 않는다는 점이 중요하다 — 업스트림 태스크로부터 XCom을 통해 전체 경로를 그대로 받아온다. 이 덕분에 태스크가 이전 단계의 로직으로부터 분리(decoupled)되고 이식 가능(portable)해진다.