DAG 작성하기 - 1

DAG란?

DAG(Directed Acyclic Graph) 방향성 비순환 그래프의 약자로 원하는 데이터를 얻긱 위해 실행되는 여러 테스크 또는 동작의 흐름으로 화살표의 방향성의 끝점을 포함하되 반복이나 순환을 허용하지 않습니다. 그래서 아래 내용을 바탕으로 보면

'download_lauches' task가 먼저 실행이 되며, 해당 task가 정상적으로 실행되어야 'get_pictures' task가 실행되며 위와 같이 순차적으로 task들이 실행이 되며, 사용자는 이러한 종속성에 맞게 DAG안에 task들을 다양한 Operator들을 통해 정의하게 되며 Python, Yaml, Bash 등 여러 가지 언어가 사용 가능하므로 복잡한 task들의 조합을 통해 효율적인 데이터 파이프라인을 구성할 수 있습니다.

 

위의 예시로 작성한 DAG를 자세히 살펴 보겠습니다.

 

TASK

1. dag - download_rocket_launches

: DAG 작성에 필요한 library Import

: DAG 선언

2. task - download_launches
: BashOperator를 통해 API의 데이터 추출 및 저장
: URL - https://ll.thespacedevs.com/2.0.0/launch/upcoming

3. task - get_pictures
: PythonOperator를 통해 추출한 JSON 데이터에서 우주선 Image의 URL을 가져와 Image 다운로드 및 저장
: 복잡한 작업 처리를 위해 Python으로 작성한 함수 생성(def _get_pictures)
: PythonOperator를 통해 작성한 함수 연결

4. task - notify & DAG 선언

: BashOperator를 연결하여 작업 완료 알림

: 작성한 Task 들의 순환 구조 선언

 

작업 실행

스케줄러 재 시작시 작성한 DAG 등록 상태 확인 및 작업 트리거

결과 파일 확인

 

테스트에 사용된 코드는 아래 URL에서 다운로드 가능합니다.

URL: https://github.com/usheep91/Airflow

 

 

참조: Apache Airflow 기반의 데이터 파이프라인