Airflow 간단 정리

JunHo Lee·2024년 1월 28일

airflow

1. Airflow란?

Airbnb(에어비앤비)에서 만든 배치 중심 워크플로우를 구축, 실행 및 모니터링 하기 위한 오픈 소스 플랫폼
- 배치 : 프로그래밍에서 '배치(batch)'는 일련의 작업들을 모아서 한 번에 처리하는 것
  1. 예시 1) 데이터베이스 작업에서 여러 개의 쿼리를 모아서 한 번에 실행
  2. 예시 2) 머신러닝에서는 전체 데이터셋을 여러 개의 작은 그룹(배치)으로 나누어 학습을 진행하는 방식으로 사용
- 워크플로우 : 의존성으로 연결된 작업들의 집합
  1. 일련의 작업들이 어떻게 연결되어 실행되어야 하는지를 정의한 것
  2. Airflow에서는 DAG(Directed Acyclic Graph)라는 개념으로 표현
- 즉, 일련을 일들을 순차적으로 진행 시켜주는 프로젝트 관리자 역할
- 기능
  - 실패 후 자동으로 작업 재실행
  - 종속성 기능(다른 작업 완료 후 시작)
  - 모니터링(UI제공)
  - DAG 실패에 대한 경고
  - 등등
Airflow의 구조
- Webserver
  1. Airflow의 로그를 보여주거나 스케쥴러(Scheduler)에 의해 생선된 DAG 목록, Task 상태 등을 시각화
  2. 즉, UI를 통해 사용자에게 시각적으로 정보를 제공해주는 요소
- Scheduler
  1. Airflow로 할당된 작업들을 스케줄링 해주는 component
  2. 스케줄 설정해둔 workflow를 실행하기 위해 Executor에게 task를 제공
- Executor
  1. 실행중인 task를 handling하는 component
  2. default 설치시에는 scheduler에 있는 모든 것들을 다 실행시키지만, production 수준에서의 executor는 worker에게 task를 push
- Workers
  1. 실제 task를 실행하는 주체자
- Database
  1. Airflow에 있는 DAG, Task 등의 metadata를 저장하고 관리
- DAG(Directed Acyclic Graph)
  1. DAG는 비순환 그래프로써 순환하는 싸이클이 없는 그래프
    1. 노드와 노드가 단방향으로 연결되어 있어 그 노드로 향하게 되면 돌아오지 않음
  2. DAG를 이용해 Workflow를 구성하여 어떤 순서로 task를 실행시킬 것인지 dependency를 어떻게 표현할 것인지 등을 설정

작업(Task)을 방향 순서대로 처리 (녹색 네모 : 작업(Task))
수집 → 분석 → 무결성 체크 → (에러 없다면) 저장 → 출력 / (에러 발생 시에는 검출해서 에러 체크)

2. 스케줄링

DAG의 schedule_interval 인수를 설정
- 프리셋 사용
  - @once 1회만 실행하도록 스케줄
  - @hourly 매시간 변경 시 1회 실행
  - @daily 매일 자정에 1회 실행
  - @weekly 매주 일요일 자정에 1회 실행
  - @monthly 매월 1일 자정에 1회 실행
  - @yearly 매년 1월 1일 자정에 1회 실행
- timedelta 사용
  - 예시 : schedule_interval=dt.timedelta(days=3) ← 3일에 한 번
  - 분은 timedelta(minutes=10), 시간은 timedelta(hours=2)
- cron 시간 설정 사용 (cron : Unix 기반 OS 스케줄러)
  분(0-59)　시간(0-23)　일(1-31)　월(1-12)　요일(0-7)
  - 매분 실행
    - * * * * *
  - 매주 금요일 오전 5시 45분에 실행
    - 45 5 * * 5
  - 매일 매시간 0분, 20분, 40분에 실행
    - 0,20,40 * * * *
  - 매일 1시 0분부터 30분까지 매분 실행
    - 0-30 1 * * *
  - 매 10분마다 실행
    - */10 * * * *
  - 5일에서 6일까지 2시,3시,4시에 매 10분마다 실행
    - */10 2,3,4 5-6 * *

3. Airflow API

Airflow 2 부터 쓸만한 REST API를 제공 (2020.12)
특징
- Swagger, redoc을 제공
  - Airflow web UI에서도 Swagger 접근 가능
- 인증을 활용한 보안 (Airflow 2 이상)
- 모든 Airflow 자원에 대해서 CRUD를 지원
기능
- DAG 실행 및 작업 정보 가져오기
- DAG 업데이트 및 삭제
- Airflow 구성 가져오기
- 연결 추가 및 삭제
- 사용자 관리
- 등등
API Endpoints 일부 소개
- DAG
  - GET /dags
    - 데이터베이스에 DAG list를 반환
    - 특정 패턴의 DAG와 일치하도록 dag_id_pattern을 설정 가능
  - PATCH /dags
    - DAG 업데이트
    - UpdateMask를 사용하여 주어진 dag_id_pattern의 DAG를 업데이트
    - 이 엔드포인트를 사용하면 ~를 dag_id_pattern으로 지정하여 모든 DAG를 업데이트할 수 있습니다. 버전 2.3.0의 새로운 기능
  - DELETE /dags
    - 완료된 DAG 실행 및 작업을 포함하여 DAG와 관련된 모든 메타데이터를 삭제
    - 로그는 삭제하지 않음
    - 이 작업은 실행 취소할 수 없음
- DAG runs
  - GET /dag/{dag-id}/dgaRuns
    - dag_id로 모든 DAG에 대한 DAG 실행을 검색
  - POST /dags/{dag-id}/dagRuns
    - dag_id로 DAG 실행
  - POST /dags/{dag-id}/clearTaskInstances
    - 지정된 날짜 범위의 DAG와 연결된 작업 인스턴스 집합 삭제
    - 실패한 실행만 삭제할 수도 있음
- Monitoring
  - GET /health
    - Airflow의 메타데이터베이스 및 스케줄러의 상태 반환
      - 여기에는 메타데이터베이스 및 스케줄러의 마지막 하트비트에 대한 정보가 포함