1. 병렬 처리병렬 처리: 여러 개의 작업을 나눠서 처리# 처리할 지역 목록regions = ['서울', '대구', '인천', '광주', '대전', '욼산', '세종']# 각 지역마다 자동으로 작업 생성parallel_task = []for region in regions: task = PythonOperator( task_id=f'process_{region}', python_callable=process_region_data, op_args=[region] ) parallel_task.append(task)# 최종 작업summary_task = PythonOperator( task_id='create_summary', python_cal..

1. TaskGroup을 활용한 복잡한 워크플로우 설계TaskGroup: 관련된 작업을 그룹화하는 것Dynamic DAG를 활용해서 TaskGroup을 여러 개 만들 수 있음# 처리해야 할 데이터 목록data_list = ['고객', '주문', '상품', '리뷰']# 각 데이터마다 자동으로 TaskGroup 만들기for data_name in data_list: with TaskGroup(f'{data_name}_처리') as group: # 1단계: 데이터 가져오기 get_data = PythonOperator(task_id=f'{data_name}_가져오기') # 2단계: 데이터 정리하기 clean_data = Pyt..

1. Dynamic DAG 생성 패턴Dynamic DAG: 설정파일이나 DB, 외부 API로부터 메타데이터를 읽어와서 동적으로 DAG를 생성하는 것 (DAG Factory)예시 코드from airflow import DAGfrom airflow.operators.python import PythonOperatorfrom datetime import datetime# 실제 작업 함수def process_data(): print("데이터 처리 중...") # 실제 데이터 처리 로직# 데이터베이스에서 설정 가져오기 (간단히)def get_configs_from_db(): # SQL: SELECT name, schedule FROM pipelines WHERE active = true re..

1. Airflow 주요 컨포넌트Webserver(사용자 인터페이스, 모니터링) ↔ Scheduler(워크플로우 엔진, 스케줄링) ↔ Executor(실행 엔진, 리소스 관리) ↔ Metadata DB(정보 저장소)사용자가 DAG 실행 요청 → Webserver가 Scheduler에 스케줄링 요청 전달 → Scheduler가 DAG 정보를 저장하고 DB에 실행가능한 Task 조회 후 반환 받기 → Scheduler가 Executor에게 Task 실행 지시한 후 Executor가 Worker에게 작업 배정 → 완료 후 Executor가 DB에 저장하게 → UI 업데이트2. Webserver 특징역할: 시각화, 실시간 상태 동기화, 메타데이터 관리(쿼리, 로그)Flask 사용, WSGI 서버 사용하는게 좋음..

1. 기본 용어DAG: 작업(Task) 간의 실행 순서나 의존성을 설정하는 방법Task: 실제 수행되는 작업 단위Operator: Task를 수행하는 실행기Scheduler: 정의된 DAG를 주기적으로 실행하는 트리거Executor: Task가 실행되는 엔진 역할 수행2. Airflow 장점작업 의존성 관리분기 처리재시도/에러 처리모니터링3. Airflow 단점자원 소모실시간 처리에 부적합의존성 관리 주의하기플러그인 연동(버전 관리) 주의하기 ✍️ 출처: [인프런] 토스 시니어 개발자와 함께하는 Data Workflow Management 기반의 대용량 데이터 처리 설계 패턴