Just Fighting
[MLOps] AirFlow 설치 및 실행하기 본문
728x90
< 설치하기 >
우선 airflow를 설치하기 위해서는 docker가 설치되어 있어야 한다.
나는 맥북에 docker와 airflow를 설치했으며, 아래 링크의 공식 홈페이지의 설치 방법을 따라 설치했다.
https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html
Running Airflow in Docker — Airflow 3.1.6 Documentation
airflow.apache.org
docker-compose란?
여러 개의 컨테이너로 이루어진 서비스(다중 컨테이너 애플리케이션)를
하나의 설정파일로 정의하고 관리할 수 있게 도와주는 도구이다.
docker는 docker run을 컨테이너 개수만큼 해줘야하지만,
docker compose는 docker-compose.yaml파일에 적어두고 실행시키면 됨.
명령어 : docker compose up / docker compose down
< 실행하기 >
DAG 생성 (EmptyOperator)
DAG파일을 생성해서 앞서 생성한 dags폴더에 넣어주면 된다.
나는 예시 DAG파일을 cursor의 도움을 받아 만들었다.
emptyOperator를 이용해 3개의 task를 만들었다.
# test_empty_operator_dag.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.empty import EmptyOperator
# DAG 기본 인자 정의
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}
# DAG 정의
with DAG(
'test_empty_operator_dag',
default_args=default_args,
description='EmptyOperator를 사용하는 테스트 DAG',
schedule=timedelta(days=1),
start_date=datetime(2026, 1, 27),
catchup=False,
tags=['test', 'empty_operator'],
) as dag:
# EmptyOperator 작업들 정의
start = EmptyOperator(
task_id='start',
)
task_1 = EmptyOperator(
task_id='task_1',
)
task_2 = EmptyOperator(
task_id='task_2',
)
task_3 = EmptyOperator(
task_id='task_3',
)
end = EmptyOperator(
task_id='end',
)
# 작업 간 의존성 정의
start >> [task_1, task_2] >> task_3 >> end
localhost:8080으로 들어가면 AirFlow 웹서버에 접속할 수 있다.
파이썬 파일을 dag폴더에 넣고, 조금 기다리면 dag항목에 새로 넣은 dag가 뜬다.
웹으로 실행도 시킬 수 있, 현재 상태, 실행 이력 등 다양한 정보를 확인할 수 있다.
PythonOperator 사용하기
파이썬 모듈을 불러와서 pythonOperator를 사용해봤다.
# dags/test_plugin_dag.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
import sys
import os
# plugins 폴더를 Python 경로에 추가
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'plugins'))
# plugins 모듈 import
from print_example import print_greeting, print_info, print_multiple_lines
# DAG 기본 인자 정의
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}
# DAG 정의
with DAG(
'test_plugin_dag',
default_args=default_args,
description='plugins 모듈을 불러와서 실행하는 테스트 DAG',
schedule=timedelta(days=1),
start_date=datetime(2026, 1, 27),
catchup=False,
tags=['test', 'plugin', 'print'],
) as dag:
# print_greeting 함수를 실행하는 작업
task_greeting = PythonOperator(
task_id='print_greeting',
python_callable=print_greeting,
op_kwargs={'name': 'Airflow User'},
)
# print_info 함수를 실행하는 작업
task_info = PythonOperator(
task_id='print_info',
python_callable=print_info,
)
# print_multiple_lines 함수를 실행하는 작업
task_multiple = PythonOperator(
task_id='print_multiple_lines',
python_callable=print_multiple_lines,
)
# 작업 간 의존성 정의
task_greeting >> task_info >> task_multiple
사용할 파이썬 모듈은 dags폴더와 함께 생성했던 plugins파일에 넣었다.
만약 다른 폴더를 plugins파일 대신에 사용하고 싶다면 초반에 다운받은 docker-compose.yaml파일의 값을 변경하면 된다.
# plugins/print_example.py
"""
Print 예문을 포함한 모듈
"""
def print_greeting(name: str = "Airflow"):
"""
인사말을 출력하는 함수
Args:
name: 인사할 대상의 이름 (기본값: "Airflow")
"""
print(f"안녕하세요, {name}님!")
print("이것은 plugins 폴더에서 가져온 모듈입니다.")
return f"Greeting printed for {name}"
def print_info():
"""
정보를 출력하는 함수
"""
print("=" * 50)
print("Airflow DAG 실행 정보")
print("=" * 50)
print(f"실행 시간: {__import__('datetime').datetime.now()}")
print("모듈 위치: plugins/print_example.py")
print("=" * 50)
return "Info printed successfully"
def print_multiple_lines():
"""
여러 줄을 출력하는 함수
"""
messages = [
"첫 번째 메시지",
"두 번째 메시지",
"세 번째 메시지",
"모든 메시지 출력 완료!"
]
for i, msg in enumerate(messages, 1):
print(f"[{i}] {msg}")
return f"Printed {len(messages)} messages"
TaskGroup 만들어보기
TaskGroup은 Task를 하나의 논리 단위로 묶어주는 UI 상의 기능이다.
시각적 분류일 뿐, 실행 로직에는 영향을 주진 않는다.
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.utils.task_group import TaskGroup
# DAG 기본 인자 정의
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}
# DAG 정의
with DAG(
'test_taskgroup_dag',
default_args=default_args,
description='TaskGroup을 사용하는 테스트 DAG',
schedule=timedelta(days=1),
start_date=datetime(2026, 1, 27),
catchup=False,
tags=['test', 'taskgroup'],
) as dag:
# 시작 작업
start = EmptyOperator(
task_id='start',
)
# 데이터 수집 그룹
with TaskGroup(group_id='data_collection') as data_collection_group:
collect_data_1 = EmptyOperator(
task_id='collect_data_1',
)
collect_data_2 = EmptyOperator(
task_id='collect_data_2',
)
collect_data_3 = EmptyOperator(
task_id='collect_data_3',
)
# 그룹 내부 의존성
collect_data_1 >> collect_data_2 >> collect_data_3
# 데이터 처리 그룹
with TaskGroup(group_id='data_processing') as data_processing_group:
process_data_1 = EmptyOperator(
task_id='process_data_1',
)
process_data_2 = EmptyOperator(
task_id='process_data_2',
)
# 병렬 처리 작업
with TaskGroup(group_id='parallel_processing') as parallel_group:
parallel_task_1 = EmptyOperator(
task_id='parallel_task_1',
)
parallel_task_2 = EmptyOperator(
task_id='parallel_task_2',
)
parallel_task_3 = EmptyOperator(
task_id='parallel_task_3',
)
# 처리 그룹 내부 의존성
process_data_1 >> parallel_group >> process_data_2
# 데이터 검증 그룹
with TaskGroup(group_id='data_validation') as validation_group:
validate_data_1 = EmptyOperator(
task_id='validate_data_1',
)
validate_data_2 = EmptyOperator(
task_id='validate_data_2',
)
# 검증 그룹 내부 의존성
validate_data_1 >> validate_data_2
# 종료 작업
end = EmptyOperator(
task_id='end',
)
# 전체 작업 흐름 정의
start >> data_collection_group >> data_processing_group >> validation_group >> end
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.utils.task_group import TaskGroup
# plugins 폴더에서 config 파일 import
from same_task_structure_config import COMMON_TASK_STRUCTURE, TASKGROUP_CONFIGS
# DAG 기본 인자 정의
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}
# DAG 정의
with DAG(
'test_same_task_structure_dag',
default_args=default_args,
description='동일한 작업 구조를 가진 여러 TaskGroup을 생성하는 DAG',
schedule=timedelta(days=1),
start_date=datetime(2026, 1, 27),
catchup=False,
tags=['test', 'taskgroup', 'same_structure', 'config'],
) as dag:
# 시작 작업
start = EmptyOperator(
task_id='start',
)
# 종료 작업
end = EmptyOperator(
task_id='end',
)
# 동일한 작업 구조를 가진 TaskGroup들을 for문으로 생성
for group_config in TASKGROUP_CONFIGS:
with TaskGroup(group_id=group_config['group_id']) as taskgroup:
# 공통 작업 구조로 Task들을 생성
tasks = {}
for task_config in COMMON_TASK_STRUCTURE['tasks']:
# 각 그룹의 설정을 사용하여 Task 생성
# 실제로는 group_config의 파라미터를 사용할 수 있음
tasks[task_config['task_id']] = EmptyOperator(
task_id=task_config['task_id'],
)
# 공통 작업 구조의 의존성을 for문으로 설정
for dep in COMMON_TASK_STRUCTURE['dependencies']:
upstream_task = tasks[dep[0]]
downstream_task = tasks[dep[1]]
upstream_task >> downstream_task
# TaskGroup 실행 순서(의존성)를 for문 안에서 설정
start >> taskgroup >> end
728x90
Comments