[Airflow] 데이터를 수집해서 결과 출력하는 DAG 작성

현기홍·2023년 11월 13일

Airflow

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

데이터 엔지니어링에서 많이 사용함. → 일련의 작업 흐름을 관리해주는 애플리케이션

ETL -데이터를 추출하고 변환하고 적재하는 작업

Airflow를 이용해서 많이 하는 작업이 데이터 수집 및 전처리와 저장 작업.

사용자 → Ariflow Web Server(DAG 실행 상태와 모니터링함)

→ dags(파이썬으로 Airflow DAG 파일을 작성) —> Airflow Scheduler —> 태스트 예약 → Airflow Worker가 예약된 작업을 예약된 이벤트가 발생하면 수행하고 그 결과를 Airflow 메타스토어에 저장하고 Web Server를 이용해서 출력

구성 요소

스케줄러 DAG(Directed Acyclic Graph - 비순환 그래프)

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

사용자가 DAG 워크플로를 작성하면 스케줄러는 DAG 파일을 분석하고 각 DAG의 태스크 의존성 및 예약 주기를 확인.

스케줄러는 마지막 DAG까지 내용을 확인한 후 DAG의 예약 주기가 경과했느지 확인해서 예약 주기가 현재 시간 이전이라면 실행되도록 예약함

kafka 게시 시작 - 1, 2, 3 (queue에 저장)

구독 시작 - 메시지 큐의 모든 내용을 가져옴

리눅스 명령어인 cron보다 더 간편하게 사용할 수 있는 것이 Airflow → DX분야에서 많이 사용함.

Airflow → Batch → Processing

예약된 각 태스크에 대해 스케줄러는 해당 태스크의 읜존성을 확인하고 의존성 태스크가 완료되지 않았다면 실행 대기열에 추가

워커

실행 대기열의 작업을 실행하는 요소

실행은 병렬로 수행되고 실행 결과는 지속적으로 추적됨

모든 결과는 메타스토어로 전달해서 사용자가 웹 인터페이스를 통해서 확인할 수 있도록 한다

웹 서버

스케줄러에서 분석한 DAG를 시각화하고 DAG 실행과 결과를 확인할 수 있도록 인터페이스를 제공한다.

모니터링과 실패 처리

웹 인터페이스를 통해서 모니터링 할 수 있고 실패한 태스크를 복구할 수 있음.

점진적 로딩과 백필(지난 날짜를 기준으로 재처리하는 것)

스케줄링을 할 때 특정 시점에 트리거 할 수 있을 뿐 아니라 최종 시점과 예상되는 다음 스케줄 주기를 상세하게 설정하는 것이 가능함.

→ 데이터 파이프라인을 구축할 때 많이 사용함

파이프라인을 이용하면 각 DAG 매번 전체 데이터 세트를 다시 처리하지 않고 해당 시간에 대한 데이터만 처리하는 것이 가능하다.

시간과 비용을 줄일 수 있다.

AIrflow는 배치 지향 데이터 파이프라인을 구축하는데 적합하다.

파이썬으로 DAG를 작성하기 때문에 파이썬으로 가능한 모든 작업을 파이프라인으로 만드는 것이 가능하다

다양한 유형의 데이터베이스나 클라우드 서비스 등과 통합이 가능한 애드온이 존재함

스케줄링 방법이 다양

백필 기능을 제공

모니터링을 위한 Web interface를 제공

오픈 소스이지만 특정 Vendor에서는 관리형 Airflow 제공

적합하지 않은 경우

스트리밍(실시간 데이터 처리) 워크플로 및 해당 파이프라인 처리에 적합하지 않을 수 있다.
추가 및 삭제 태스크가 빈번한 동적 파이프라인에는 적합하지 않을 수 있음
파이썬 언어를 알아야 한다
파이썬 코드로 DAG를 작성하기 때문에 파이프라인 규모가 커지면 굉장히 복잡해짐

데이터 수집 샘플

데이터를 가져올 API ll.thespacedevs.com

수행할 작업

API를 가져와서 파싱한 후 이미지를 다운로드 받아서 저장.
API 를 다운로드 받아서 저장한 후 이미지 URL을 이용해서 이미지를 다운로드 받아서 저장하고 echo로 결과를 출력
작업을 작성할 때는 파이썬 코드로 작성하는데 작업 자체는 파이썬 대신에 배치 스크립트(linux)를 사용

DAG 파일 작성

# download_image.py

import json
import pathlib

import airflow.utils.dates
import requests
import requests.exceptions as requests_exceptions
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

# DAG 인스턴스 생성 - 모든 워크플로의 시작점
# id와 description은 마음대로 작성 가능

dag = DAG(
    dag_id = 'download_image',
    description = "Download Image",
    start_date = airflow.utils.dates.days_ago(14),
    schedule_interval = None
)

# 리눅스 작업을 생성
# 웹 서버 API 데이터는 되도록이면 curl을 이용하는 것이 편함.

download_launches = BashOperator(
    task_id = 'download_launches',
    bash_command = "curl -o /tmp/launches.json -L 'https://ll.thespacedevs.com/2.0.0/launch/upcoming'", # noqa: E501
    dag = dag
)

def _get_pictures():
    pathlib.Path('/tmp/images').mkdir(parents=True, exist_ok=True)

    with open('/tmp/launches.json') as f:
        launches = json.load(f)
        image_urls = [launch['image'] for launch in launches['results']]
        for image_url in image_urls:
            try:
                response = requests.get(image_url)
                image_filename = image_url.split('/')[-1]
                target_file = f"/tmp/images/{image_filename}"
                with open(target_file, "wb") as f:
                    f.write(response.content)
                print(f"Downloaded {image_url} to {target_file}")
            except requests_exceptions.MissingSchema:
                print(f"{image_url} appears to be an invalid URL")
            except requests_exceptions.ConnectionError:
                print(f'Could not connect to {image_url}')

#DAG에서 파이썬 함수를 호출 
get_pictures = PythonOperator(
    task_id="get_pictures", python_callable=_get_pictures, dag=dag
)

notify = BashOperator(
    task_id = "notify",
    bash_command='echo "There are now $(ls /tmp/images/ | wc -l) images."',
    dag=dag
)

download_launches >> get_pictures >> notify