Chapter1. Apache Airflow 살펴보기

1.1 데이터 파이프라인

데이터 파이프라인 : 원하는 결과(특정 형태의 데이터, 모델)를 얻기 위해 여러 태스크 또는 동작으로 구성된 워크플로우

• 각 테스크는 정해진 순서대로 진행되어야 함.

  • 데이터를 가져오기 전 변환을 시도하면 의미가 없는 태스크가 됨.

이러한 태스크의 순서, 테스크의 의존성을 표현한 것이 데이터 파이프라인

데이터 파이프라인은 방향성을 가지며 반복이나 순환을 허용하지 않는 방향성 비순환 그래프(DAG) 로 구성되어 있음

5p. figure1.3 참고

1.1.2 파이프라인 그래프 실행

1. 그래프 안에 태스크는 각각 open 상태이며 다음과 같은 단계로 실행

• 각각의 화살표 끝점은 테스크를 향하고 다음 태스크로 향하기 전 이전 태스크가 완료되었는지 확인

• 태스크가 완료되면 다음에 실행해야 할 태스크를 대기열에 추가

1. 실행 대기열에 있는 태스크를 실행하고 수행이 완료되면 완료 표시를 함.

2. 그래프의 모든 태스크가 완료될때까지 1단계로 돌아감

7p. figure1.4 참고

1.1.3 그래프 파이프라인 vs 절차적 스크립트 파이프라인 비교

절차적 스크립트 파이프라인(일반적인 ML과정)

8p. figure1.5 참고

이 과정을 아래와 같이 그래프 기반으로 표현하면 각 태스크를 노드로 생성하고, 태스크 간의 데이터 의존성을 화살표 끝점으로 연결하여 표현할 수 있음.

그래프 파이프라인

8p. figure1.6 참고

첫번째 단계를 독립적인 두개의 태스크로 분리되어 개별적인 태스크를 명확하게 파악할 수 있음.

그래프 기반 표현의 장점

• 전체 작업을 하나의 모놀리식(단일) 스크립트로 구성되는 것이 아니라 파이프라인을 작은 점진적인 태스크로 명확하게 분리 할 수 있음.

  • 모놀리식의 경우 중간 태스크가 실패하면 전체 스크립트를 재실행해야 하는 문제가 발생

• 실패한 태스크(+ 해당 태스크 이후 태스크만) 실행하면 되어 효율적으로 구성할 수 있

1.1.4 워크플로 매니저 종류

9p. Table 1.1 참고

1.2 Airflow 소개

• Airflow™ is a platform created by the community to programmatically author, schedule and monitor workflows.

• 워크플로를 개발하고 모니터링하고 위한 오픈 소스 솔루션

1.2.1 파이썬 코드로 유연한 파이프라인 정의

• 파이썬 코드로 파이프라인, 워크플로 태스크를 방향성 비순환 그래프(DAG)로 정의

• DAG 파일은 주어진 DAG에 대한 태스크 집합과 태스크 간의 의존성을 기술하고, Airflow가 DAG구조를 식별하기 위해 코드를 파싱

• 또, DAG 파일에는 Airflow 실행 방법, 실행 시간을 정의

• 파이썬 코드로 작성하여 외부 데이터베이스, 빅데이터 기술(ML, Spark), 클라우드 서비스 등 다양한 시스템에서 실행할 수 있도록 확장성이 좋음

1.2.2 파이프라인 스케줄링 및 실행

• 파이프라인을 언제 실행할 것인지, 각각의 DAG의 실행 주기를 정의할 수 있음

• 매시간, 매일, 매주 DAG를 실행하거나 복잡한 스케줄을 사용할 수 있음

Airflow 구성

• Airflow 스케줄러

  • DAG를 분석하고 현재 시점에서 DAG 스케줄이 지난 경우 Airflow 워커에 DAG의 태스크를 예약

• Airflow 워커

  • 예약된 태스크를 선택하고 실행

• Airflow 웹서버

  • 스케줄러에서 분석한 DAG를 시각화하고 DAG 실행과 결과를 확인할 수 있는 주요 인터페이스를 제공

12p figure 1.8 참고

13p figure 1.9 참고

Airflow의 전체 실행과정을 이해하기보다는 대략적인 흐름만 파악 DS 영역에서는 DAG 파일 작성, 웹 인터페이스로 태스크 실행 결과 모니터링이 주된 역할

1.2.3 모니터링 실패처리

• Airflow 핵심 기능 중 하나

• DAG를 확인하고 실행 결과에 대한 모니터링 제공

• 태스크 실패 시 재시도(재실행 시간의 간격을 설정할 수 있음) 하거나 오류 발생 시 태스크를 복구할 수 있음

• 실패한 태스크를 사용자에게 알리고 실패한 태스크의 로그를 확인할 수 있어 디버깅에 장점이 있음

1.2.4 점진적 로딩 및 백필

• Airflow 핵심 기능 중 하나

• Airflow는 데이터 파이프라인을 점진적으로 실행

  • DAG는 매번 전체 데이터 세트를 다시 처리할 필요 없이 해당 시간 슬롯에 대한 데이터만 처리

  • 대규모 데이터 세트 처리해야 할 경우 기존 결과에 대한 태스크 전체를 다시 수행하는 것을 방지해 비용과 시간을 절약

• 스케줄 주기 + 백필 기능과 결합하여 새로 생성한 DAG를 과거 시점 및 기간에 대해 실행 가능

  • 특정 기간에 대해 DAG를 실행해 새로운 데이터를 손쉽게 생성 또는 백필 할 수 있음

Backfill : 데이터 파이프라인을 운영하며, 이미 지난 날짜를 기준으로 재처리 하는 일

• 버그가 있거나 어떤 이유로 로직이 변경되었을 때 전체 데이터를 새로 말아주어야 할 때

• 컬럼 등의 메타 데이터가 변경되었을 때 이를 반영하기 위한 append 성의 작업이 필요할 때 이외에도 과거의 데이터를 재처리하고자 하는 니즈가 있다면 백필을 먼저 생각하면 됨

참고. Airflow 체험

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator

# DAG 정의
default_args = {
    'owner': 'your_name',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'simple_airflow_dag',
    default_args=default_args,
    description='A simple Airflow DAG',
    schedule_interval=timedelta(seconds=10),  # DAG 실행 주기 (10초마다 한 번)
)

# 첫 번째 작업 정의
task1 = BashOperator(
    task_id='print_hello',
    bash_command='echo "Hello Airflow"',
    dag=dag,
)

# 두 번째 작업 정의
task2 = BashOperator(
    task_id='print_world',
    bash_command='echo "World!"',
    dag=dag,
)

# 작업 간 의존성 정의
task1 >> task2

if __name__ == "__main__":
    dag.cli()