[Airflow] 개념정리 01.

Apache Airflow

• 에어비엔비에서 만든 workflow 관리 툴
• workflow는 "일련의 작업의 흐름"
  • ETL (데이터를 추출 -> 변환 -> 적재)하는 workflow를 관리해주는 툴이 Airflow
  • 이런 workflow management 도구는 하둡 에코시스템의 우지(oozie), 루이지(luigi)같은 솔루션이 있음.
• Scheduler가 DAG directory의 작업을 가져와서 Workers에서 실행하는 형태.
• "배치 지향 데이터 파이프라인 구현"을 위해 특화된 주요 기능을 가지고 있음.
  • Python기반으로 만들어졌음. 커스텀 파이프 라인 구성 가능함.
  • 쉽게 확장 가능하고 다양한 시스템과 통합이 가능함.
    • 다양한 유형의 데이터 베이스, 클라우드 서비스 등과 통합할 수 있는 수 많은 애드온이 존재함.
  • 스케줄링 기법 : 정기적으로 실행하고 점진적(증분)처리를 통해, 전체 파이프라인을 재실행할 필요 없는 효율적인 파이프라인 구축 가능.
    • 백필 기능을 통해 과거 데이터를 손쉽게 처리할 수 있기 때문에 코드를 변경한 후 재생성이 필요한 데이터 재처리가 가능함.
  • 웹 인터페이스를 통해 파이프라인 실행 결과를 모니터링 할 수 있고 오류를 디버깅하기 편리한 뷰를 제공함.
  • 오픈 소스.
    • 특정 벤터에 종속되지 않고 사용가능함.
• Airflow 콘솔이 따로 존재해 Task 관리를 서버에서 들어가 관리하지 않아도 되고, 각 작업별 시간이 나오기 떄문에 bottleneck(병목현상)을 찾을 때에도 유용함.

 

Airflow Architecture

• Scheduler - Airflow의 DAG와 작업들을 모니터링하고 실행 순서와 상태 관리
  • 작업 기준이 충족되는지 여부를 확인.
  • 종속작업의 성공여부와 예약간격의 실행가능 여부, 실행조건의 충족여부 확인.
  • 충족 여부가 DB에 기록되면, Task들이 worker에서 선택되어 작업을 실행함.
• Workers - Airflow의 작업을 실행하는 공간
• Metadata Database - Airflow에서 실행할 작업에 관한 정보들을 저장
• Webserver - Airflow의 user interface 제공
  • workflow 상태 표시하고 실행, 재시작, 수동 조작, 로그 확인 등 가능
• DAG directory - Airflow에서 실행할 작업들을 파이프라인 형태로 저장

 

Airflow 사용이 적합하지 않은 경우.

• airflow는 반복적이거나 배치 태스크를 실행하는 기능에 초점이 맞춰져 있기 때문에, 스트리밍(실시간 데이터 처리)워크 플로 및 해당 파이프라인 처리에 적합하지 않음.
• 추가 및 삭제 태스크가 빈번한 동적 파이프라인의 경우 적합하지 않음.
  • 동적 태스크를 구현할 수 있으나, 웹 인터페이스는 DAG의 가장 최근 실행 버전에 대한 정의만 표현함
  • 실행되는 동안 구조가 변경되지 않는 파이프라인에 적합함.

 

---

DAG 

비순환 그래프로써 순환하는 싸이클이 없는 그래프

• 즉, 노드와 노드가 단방향으로 연결되어 있어 그 노드로 향하게 되면 돌아오지 않는 특성을 가지고 있음.
• Airflow에서는 Task의 연결관계를 DAG로 관리하고, Webserver를 통해서도 DAG 구조를 시각적으로 확인할 수 있음.

---

Airflow Senseors

• 특정 조건이 참이 되길 기다리는 연산자
  • "airflow.sensors.base_sensor_operator" 클래스에서 파생.
  • parameter
    • mode : 센서 상태확인
      • mode="poke" : 완료시까지 계속 확인
      • mode="reschedule" : 다른 슬롯을 사용할 수 있을때까지 기다림
    • poke_interval : pork 모드에서 airflow에 조선을 확인하는 빈도를 알려줌.
    • timeout : 작업실패로 표시하기 전에 대기하는 시간.
• file sensor
  • 파일 시스템의 특정 위치에 파일이 있는지 확인.

 

---

Task Instance

none	task가 대기열에 올라가지 않음 (종속성이 아직 충족되지 않음).
scheduled	스케줄러가 task의 종속성이 충족되고 실행되어야 한다고 결정한 상태
queued	task가 Executor에 할당되었으며 실행되기를 기다리고 있는 상태
running	task가 실행 중
success	task 성공
failed	task를 실행하는 동안 오류가 발생하여 실패
skipped	조건식, LatestOnly 등으로 인해 작업을 건너뜀
upstream_failed	upstream task가 실패했고 trigger rule이 필요한 상태
up_for_retry	task가 실패했지만 재시도 횟수가 남아 있어서 재 스케줄을 할 예정인 상태
up_for_reschedule	해당 task가 `reschedule` 모드인 sensor라고 나타내는 상태
sensing	해당 task가 Smart Sensor 라고 나타냄
removed	DAG가 실행된 이후에 task가 삭제됨

 

---

Reference

https://airflow.apache.org/docs/apache-airflow/stable/index.html

What is Airflow? — Airflow Documentation

https://airflow.apache.org/docs/apache-airflow/stable/_modules/airflow/sensors/filesystem.html#FileSensor

airflow.sensors.filesystem — Airflow Documentation