본 글은 Apache Airflow 공식 문서와 GPT를 참고하여 작성했습니다.

 

Airflow란?

developing, scheduling, monitoring workflows의 자동화를 도와주는 플랫폼이다.

Ariflow는 다음과 같은 특징들이 있다.

  • Scalable
    하나의 single process에서 distribued system까지 확장이 가능하다.
    modular arhcitecture와 message queue를 사용해서 무한한 수의 workers를 orchestration 할 수 있다.
  • Batch based workflows
    정해진 시점에 정해진 workflows를 처리하는 작업 단위를 의미한다.
    Modeling에서는 한 번에 처리할 수 있는 data 크기와 연관된 개념이지만, 여기서는 특정 시점의 작업 단위라고 생각하면 된다.

🤔 Workflow vs Pipeline
1. Pipeline
- Input을 일련의 Transformation을 거쳐 Output이 되도록 하는 흐름.
- 선형적인 구조를 가진다(A → B → C)

2. Workflow
- 더 복잡한 작업들을 관리하는 흐름. 즉 worfklow ⊃ pipeline라 볼 수 있다.
- 조건부 로직(Branching, 경로 설계), 의존성 관리(Dependency, A가 끝나야 B 시작), error handling 및 retry(특정 stage에서 에러 발생 시 다시 시도 혹은 처리), data 전처리 뿐만 아니라 서버 점검, API 호출 등 데이터 흐름과 상관 없는 업무까지 포함
- DAG 구조를 가진다.

  • Workflows as code
    Airflow pipelines는 모두 python으로 정의되며 다음과 같은 장점을 제공한다.
    • Dynamic: Pipelines을 code로 정의해서 동적 DAG generation(pipeline instantiation을 동적으로 수행)과 parameterization(workflow 실행 시 parameter를 주입)을 가능하게 하며, 실행 시점에 유연하게 worfklows를 변화시킬 수 있음
    • Extensible: built-in operators 뿐만 아니라 사용자 정의 operators를 쉽게 생성할 수 있고, env에 최적화된 abstraction level에 맞춘 libraries를 확장할 수 있다.
    • Flexible: Jinja라는 Python 용 web templating engine이 core에 내장되어 있어, customizations과 Parametrization이 가능하다.

 

그렇다면 이런 Airflow를 언제 사용하면 좋을까? 

아주 아주 단순한 예시를 생각해보면, 우리가 어떤 모델을 특정 parameter를 외부에서 전달하여 Inference 성능을 여러 번 평가하는 python 프로그램을 작성했다 가정해보자.

 

터미널에서 python eval.py --params param1 command를 입력해서 parameter 값들을 argparse로 argument를 전달하거나, bash scripts를 작성하는 방식을 떠올릴 수 있다.

 

그러나 평가해야 할 파라미터 조건이 100개로 늘어난다면 어떨까? 그 중 하나가 실패한다면? 각 실행 log들은 어떻게 기록하고 추적할까? 

 

만약 실제 운영 환경이라면 작업의 scale이 커지고 환경이 확장되기 때문에, 수동으로 제어하기 어려운 복잡한 사항들이 많아질 것이다! 이럴 때 이런 작업들을 세분화하고 관리하는 도구들이 필요할 것이라 생각한다. 

 

 

DAG

이런 자동화 platform에서 가장 핵심 개념 중 하나가 바로 DAG이다.

DAG는 directed acyclic graph(방향이 있는 비순환 그래프)라는 수학적인 개념에서 유래했지만, Airflow에서는 workflow를 정의하는 구체적인 data structure로서 사용된다.

즉, DAG는 Airflow가 workflow를 어떻게 표현하고 관리하는지를 보여주는 Model로서 workflow 실행에 필요한 모든 정보를 캡슐화한다.

  • 주요 DAG attributes
    • Schedule: workflow가 실행되어야 할 시점 및 주기
    • Tasks: workers에 의해 실행되는 discrete 작업 단위(unit),
    • Task Dependencies: tasks간의 order와 conditions
    • Callbacks: 전체 workflow 완료 시(성공/실패) 취해야 할 action
    • Additional Parameters: 다른 많은 운영 details

추후 Core Concepts 글에서 DAG attributes에 대해서 다시 작성해보겠다.^^

 

Simple Dag Code snippet

간단한 DAG 정의 후 airflow에서 어떻게 보여지는지 알아보자.

from datetime import datetime

from airflow.sdk import DAG, task
from airflow.providers.standard.operators.bash import BashOperator

# A Dag represents a workflow, a collection of tasks
with DAG(dag_id="demo", start_date=datetime(2022, 1, 1), schedule="0 0 * * *") as dag:
    # Tasks are represented as operators
    hello = BashOperator(task_id="hello", bash_command="echo hello")

    @task()
    def airflow():
        print("airflow")

    # Set dependencies between tasks
    hello >> airflow()

위의 scripts에는 다음과 같은 정보가 있다.

  1. Dag name: demo
  2. Schedule: 2022년 1월 1일에 시작해서 매일 실행된다.
  3. 2개의 tasks:
    - BashOperator: shell script를 실행 담당
    - @task decorator: Python function을 task로 정의
  4. >> operator: 2개의 tasks dependency와 execution 순서를 control

Airflow는 위의 script를 Parsing → Tasks scheduling → Executing 한다.

Web UI 상의 "demo" Dag 상태

 

이 example은 단순한 Bash command와 Python function을 사용했지만, Airflow tasks는 사실상 어떤 종류의 코드든 실행 할 수있다.

EX) Spark 작업을 수행하거나 storage buckets 간에 files를 이동시키고, 알림 email을 보내는 등의 다양한 용도로 taks를 활용할 수 있다!

 

위의 그림은 동일한 DAG가 시간이 흐름에 따라 multiple 실행되었을 때의 모습이다. (시간에 따른 execution history가 가로로 나열된 모습이다) 

오른쪽에 보이는 Grid의 각 Column은 하나의 Single DAG Run을 의미한다.

 

보통 Graph와 Grid view가 가장 많이 쓰이지만, workflows를 monitoring하거나 troubleshoot하는데 도움이 되는 Dag Overview 등 다양한 view를 제공한다.

 

Why Airflow?

Airflow는 batck workflows를 orchestrating하기 위한 platform이며 built-in operators를 제공하고 새로운 기술과의 통합이 쉽다는 것을 위에서 언급했다.

 

공식 문서에 따르면 

  1. workflows의 시작과 끝이 명확할 때
  2. 정해진 schedule대로 실행되어야 할 때 

위와 같은 상황에 사용하는 것을 추천한다. 

 

또한 Python code로 worfklows를 정의했을 때의 장점은 다음과 같다! 

  • Version control: 변경 사항을 tracking 및 이전 version을 roll back
  • Team collaboration: 여러 개발자가 동일한 codebase에서 협업 가능
  • Testing: units / integration test들을 통해서 pipeline logic validation
  • Extensibiltiy: 기존 components 활용 및 직접 빌드하여 workflows를 Customizing

이 밖에 

  • Web interface: Dag 수동 trigger, logs monitoring, task status 확인
  • Backfill: DAG 실행을 과거 시점에 대해서도 실행하는 기능. 즉, 파이프라인을 오늘 만들었더라도, 스케줄을 과거로 설정하면과거 시점의 데이터부터 처리할 수 있다. 
  • Rerun only Failed Task: cost와 time 최소화를 위해서 실패한 task에 대해서만 재실행 가능

그리고 무엇보다 OpenSource 라는 거! (고마워요 Aribnb)

 

Why not Airflow?

  • Continuously Running, Event-driven, Streaming workloads 를 처리
    유한한 batch-oriented workflows이기 때문에 반드시 끝나야 하는 job을 관리하도록 설계되었다.
    물론 CLI 나 REST API를 사용해서 DAG를 trigger할 수 있지만 24시간 내내 실행하거나, 이벤트 기반 혹은 실시간 스트리밍 처리에는 적합하지 않다. 

대신 Airflow는 종종 Apache Kafka 같은 streaming systems을 보완하는 역할을 한다. Kafka가 실시간 데이터 수집을 담당하여 저장소에 데이터를 기록하면, Airflow는 주기적으로 그 데이터를 가져와 batch 단위로 처리한다.

'DevOps > Airflow' 카테고리의 다른 글

[Airflow] Docker Compose로 Installation  (0) 2026.01.14
[Airflow] Architecture  (0) 2026.01.13

티스토리툴바