[Airflow] 병렬 처리 및 분산 처리와 대용량 재처리

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_callable=create_national_summary
)

# 병렬 처리 후 요약 생성
parallel_task >> summary_task

def process_customer_batch(start_id, end_id):
    # 고객 데이터를 배치 단위로 처리
    print(f"고객 ID {start_id}-{end_id}) 처리 시작")

# 100만 명의 고객을 10만 명씩 나눠서 처리
batch_size = 100000 # 배치 크기
total_customers = 1000000 # 전체 고객 수

batch_tasks = []

for i in range(0, total_customers, batch_size):
    start_id = i + 1
    end_id = min(i + batch_size, total_customers)

    batch_task = PythonOperator(
        task_id=f'process_batch_{start_id}-{end_id}',
        python_callable=process_customer_batch,
        op_args=[start_id, end_id] # 순서대로 start_id, end_id 인자에 매핑됨
    )
    batch_tasks.append(batch_task)

final_aggregation = PythonOperator(
    task_id='final_aggregation',
    python_callable=aggregate_all_results
)

batch_tasks >> final_aggregation

관련 매개변수
- op_args: 위치 인자 (리스트)
- op_kwargs: 키워드 인자 (딕셔너리)
    예) op_kwargs={'start_id': 1, 'end_id': 100}

• 물리적 자원의 한계를 고려하여 분산 처리가 필요함

---

2. 분산 처리

• 분산 처리: 리소스 점유를 방지하기 위해 작업을 분산해서 처리

# airflow.cfg 설정 (간단히)
executor = CeleryExecutor
celery_broker_url = redis://redis-server:6379/0

# 분산 처리용 작업 정의
def heavy_computation(data_chunk):
    # 무거운 계산 작업
    print(f"데이터 청크 처리 시작: {len(data_chunk)}건")

    # 복잡한 계산 작업 시뮬레이션
    result = sum(data_chunk) * 2

    print(f"계산 완료: 결과 = {result}")
    return result

# 분산 처리핧 작업들
distributed_tasks = []

data_chunks = [
    [1, 2, 3, 4, 5],    # 청크1
    [6, 7, 8, 9, 10],   # 청크2
    [11, 12, 13, 14, 15],   #청크3
    [16, 17, 18, 19, 20]    #청크4
]

for i, chunk in enumerate(data_chunks):
    task = PythonOperator(
        task_id=f'heavy_computation_{i}',
        python_callable=heavy_computation,
        op_args=[chunk],
        queue='heavy_computation_queue' # 특정 큐에 할당 
        # airflow celery worker --queues heavy_computation_queue 
        # => 이 Worker는 heavy_computation_queue에 할당된 작업만 처리
    )
    distributed_tasks.append(task)
    
collect_results = PythonOperator(
    task_id='collect_results',
    python_callable=collect_all_results
)

distributed_tasks >> collect_results

• pool 사용

memory_intensive_task = PythonOperator(
    task_id='process_large_dataset',
    python_callable=process_large_data,
    pool='high_memory_pool',  # 메모리 전용 풀 사용
    priority_weight=10  # 높은 우선순위
)

# CPU를 많이 사용하는 작업
cpu_intensive_task = PythonOperator(
    task_id='complex_calculation',
    python_callable=complex_math,
    pool='high_cpu_pool',     # cpu 전용 풀 사용
    priority_weight=5   # 중간 우선순위
)

# 파일 입출력 작업
io_task = PythonOperator(
    task_id='file_processing',
    python_callable=process_files,
    pool='io_intensive_pool',   # I/O 전용 풀 사용
    priority_weight=10 # 낮은 우선순위
)

# airflow pools set high_memory_pool 2 "High memory consuming tasks"
# airflow pools set high_cpu_pool 4 "CPU intensive tasks"
# airflow pools set io_intensive_pool 10 "I/O heavy tasks"

---

3.  Backfill 전략과 대용량 재처리

• 과거 데이터를 다시 재처리해야 하는 상황에서 사용됨.    예) 비지니스 로직 변경, 데이터 오류 발견
• 주의사항: 기존 데이터 백업

# 특정 하루 재처리
airflow dags backfill --start-date 2024-01-15 --end-date 2024-01-15

 

 

 

 

 

✍️ 출처: [인프런] 토스 시니어 개발자와 함께하는 Data Workflow Management 기반의 대용량 데이터 처리 설계 패턴