Ch.11 Airflow로 효율적인 데이터 파이프라인 작성하는 방법

#️⃣ [1] 깔끔한 DAG 작성

1.1. 스타일 가이드 사용

• PEP8 스타일가이드 및 구글과 같은 회사 가이드 맟추기 : 변수/클래스/함수 이름 지정 스타일등에 대한 권장 사항
• pylint와 flake8과 같은 정적 코드 검사기로 코드 품질 확인
• YAPF, Black과 같은 파이썬 코드 포맷터 사용하여 공통 포맷 적용함
• Airflow 코드 스타일 규칙에 맞추기
  • DAG 정의 스타일 2가지 : 콘텍스트 매니저 사용 및 사용 X
  • Task 종속성 표기법

1.2. 중앙에서 자격 증명 관리

• Airflow 메타스토어에서 자격증명 가져오기
• 혹은 배포 방법에 따라 보안 정보를 외부 시스템에 저장

1.3. 구성 세부 정보 일관성 있게 지정하기

• 파일 경로나 테이블명과 같은 DAG 구성 정보는 전역변수(DAG내), 구성파일 (YAML, JSON 등), Airflow Variables를 사용하여 저장할 수 있다.
• 단 Airflow Variables에 저장하고 불러올 시 전역 범위가 아닌, Task 내에서 필요한 때에 호출하는 것이 좋다.
• ⇒ DRY(Don’t Repeat Yourself) 원칙에 따라 단일 위치에 구성 값 저장하기

1.4. DAG 구성 시 연산 부분 배제

DAG 정의 시 장기간 실행 또는 과도한 계산 수행시, DAG 파일 로드하는 문제가 있음

실제 계산이 필요한 시점에만 해당 태스크를 실행하도록 하는 것이 좋다. (태스크 내에서 호출할 것)

• (연산 예시) 자격 증명을 가져오는 훅/오퍼레이터 작성, 구성 정보를 외부 데이터 소스 또는 파일 시스템로부터 로드

1.5. Factory 함수 혹은 TaskGroup으로 공통 패턴 생성

Airflow 1.0 버전의 경우 factory 함수를 이용하여 공통 DAG 및 Task구조를 생성할 수 있다.

하지만, Airflow 2.0 부터는 TaskGroup(태스크 그룹)이 도입되며, 태스크 세트를 더 작은 그룹으로 효과적으로 그룹화하여, 시각적으로도 논리적으로도 DAG 구조를 보다 쉽게 관리하고 이해가능하다.

1.6. 대규모 수정을 위한 새로운 DAG 생성

• DAG에서 큰 변경이 발생하면, 스케줄러가 의도한 대로 동작하지 않을 수 있다. 특히 태스크를 제거하거나 이름 변경 시 UI에서 해당 태스크 기록에 액세스 불가하다
• 해당 문제를 방지하는 가장 좋은 방법은 대규모 수정이 발생하는 DAG를 복사하여 새로운 이름으로 배포하는 것이다.

---

#️⃣ [2] 재현 가능한 태스트 설계

태스크를 다른 시점에서 실행할 때에도, 태스크를 간단하게 다시 실행하고 동일한 결과를 기대할 수 있어야한다.

2.1. Task는 항상 멱등성을 가져야한다.

동일한 태스크를 여러 번 다시 실행하더라도 그 결과를 항상 동일해야한다.

2.2 Task 결과는 결정적이어야 한다.

태스크는 주어진 입력에 대해 항상 동일한 출력을 반환해야한다.

2.3 함수형 패러다임으로 Task 설계하자.

함수형 프로그래밍 패러다임에 따라 태스크를 설계하면, 멱등성 및 결정론적 태스크의 목표를 달성하여 DAG 및 태스크를 재현 가능하게 만들 수 있다.

---

#️⃣ [3] 효율적인 데이터 처리

3.1 데이터 처리량 제한하기

필요한 최소한의 데이터로 처리를 제한하여 효율적으로 처리하는 것이 좋다.(사용하지 않는 행과 열을 삭제하기)

필터링과 집계단계의 수행 순서를 앞으로 배치하여, 조인 전에 데이터 세트의 크기를 줄인다.

3.2 증분 적재 및 처리

데이터 증분 처리를 사용하여, 각 처리 실행 단계에서 수행해야 하는 처리량을 제한한다. (특히 시계열 데이터 세트가 대표적)

데이터를 (시계열 기반) 파티션으로 분할하고, 해당 파티션을 각 DAG실행헹서 개별적으로 처리한다.

• 장점 : 실행 중 오류 발생시, 실패한 부분만 다시 실행가능함

3.3 중간 단계 데이터 캐싱

태스크 중간 단계 데이터를 저장하면 각 태스크를 다른 태스크와 독립적으로 쉽게 다시 실행가능하다.

• e.g. 중간 단계를 클라우드 스토리지에 저장가능하다
• 단점 : 대규모 데이터 세트의 여러 중간 버전은 과도한 양의 스토리를 필요로 한다.
• 절충안 : 데이터의 가장 원시 버전을 항상 사용 가능하도록 보관한다.

3.4. 로컬 파일 시스템에 데이터 저장 금지

중간단계 데이터 로컬 파일 시스템에 저장하는 경우,병렬 처리 시 다운 스트림 태스크가 파일에 접근하지 못할 수도 있다.

그러므로 모든 Airflow 워커에서 동일하게 액세스할 수 있는 공유 저장소(예. 공유 클라우드 스토리지 버킷)를 사용하는 것이 좋다.

3.5. 외부/소스 시스템으로 작업 이전하기

Airflow는 원래 목적에 맞게 오케스트레이션 도구로 사용할 때가 BEST!

각 데이터 처리 부분은 더 효율적인 외부 시스템을 사용하는 것이 좋음 (e.g. Spark)

---

#️⃣ [4] 자원 관리

4.1. Pool을 이용하여 동시성 관리

많은 태스크를 병렬 실행 시, 여러 task가 동일한 자원에 접근할 수 있다. 이러한 동시성을 처리하도록 Airflow는 Pool을 제공함

• Resource Pool(리소스 풀)을 사용하여 주어진 리소스에 액세스 할 수 있는 Task 수를 제어
• 각 풀은 해당 리소스에 대한 액세스 권한을 부여하는 고정된 수의 Slot(슬롯)을 갖고 잇음
• 슬롯의 수는 리소스 풀의 동시성 정도를 정의함

4.2. SLA 및 경고를 사용하여 장기 실행 작업 탐지

• SLA(Service-Level-Argreement;서비스 수준 계약) 매커니즘: 태스크 동작 모니터링 가능
• SLA 기능으로 DAG및 TASK에 SLA 제한 시간을 효과적으로 지정 가능함
• 적용하면 Airflow는 각 태스크 실행 결과를 검사하여 태스크의 시작 및 종료 시간이 SLA에를 초과했는지 확인함→ 초과 시 경고를 Airflow 메타스토어에 저장하고, UI로도 확인 가능함 (혹은 이메일 주소로 경고 및 sla_miss_callback 매개변수로 사용자 핸들러 정의 가능함)

 

---

참고 : [책] Apache Airflow 기반의 데이터 파이프라인