[python] Apache Airflow 소개 및 실습하기(기초)

2021.03.18 21:22

졸리운_곰 조회 수:308

[python] Apache Airflow 소개 및 실습하기(기초)

Apache Airflow는 배치 스케쥴링(파이프라인) 플랫폼입니다

실행할 Task(Operator)를 정의하고 순서에 등록 & 실행 & 모니터링할 수 있습니다

DAG(Directed Acyclic Graph, 비순환 방향 그래프)로 각 배치 스케쥴이 관리됩니다

DAG하위에는 고유한 여러 Task가 존재하며 순서를 갖습니다

Task는 BashOperator, PythonOperator 등 다양한 Operator를 지원합니다

DAG(Directed Acyclic Graph) ID ├── Task ID 1 ├── Task ID 2 ├── Task ID 3

[배치 스케쥴링할 때 고려해야 할 문제들]

일부 Task가 실패했습니다. 어떻게게 인지하고 재실행할 수 있을까요?

각 Task는 선행조건과 수행시간이 다릅니다. 어떤 순서가 최적일까요?

수행시간이 일부 Task는 짧고 일부 Taks는 깁니다. 어떤 이유일까요? 로그는 쉽게 찾아 비교할 수 있을까요?

여러 서버에 Task가 존재합니다. 어떻게 스케쥴링 할 수 있을까요?

과거 일부 Task가 실행이 누락되었습니다. 과거기준으로 다시 실행할 수 있을까요?

Airflow는 이 문제에 해답을 제시합니다

Apache Airflow Documentation — Airflow Documentation

Airflow 1.10.3 Project License Quick Start Installation Tutorial How-to Guides UI / Screenshots Concepts Data Profiling Command Line Interface Scheduling & Triggers Plugins Security Time zones Experimental Rest API Integration Metrics Kubernetes Lineage Changelog FAQ Macros reference API Reference D...

airflow.apache.org

그럼 실습환경을 구축하고 실습코드를 만들어 스케쥴링해보겠습니다

1) 도커 이미지를 가져옵니다

docker pull jupyter/base-notebook:python-3.7.3

2) 도커 컨테이너를 실행합니다

docker run --name py3 -it -p 8881-8889:8881-8889 -v C:\notebooks\:/notebooks/ jupyter/base-notebook:python-3.7.3 bash

3) conda를 이용하여 필요한 모듈을 설치합니다

# conda 최신버전으로 업데이트한다 conda update -y conda # airflow 모듈을 설치한다(버전 1.10.3) conda install -y airflow # vim 모듈을 설치한다(소스코드 편집용) conda install -y vim

4) cfg(환경설정파일)과 기본데이터베이스(SQLite)를 초기화합니다 (SQLite이외 다른 DB 사용 가능)

airflow initdb

5) 사용자 환경변수에 airflow 경로를 추가하고 활성화합니다

echo 'export AIRFLOW_HOME=~/airflow' >> /home/jovyan/.profile echo 'export AIRFLOW_HOME=~/airflow' >> /home/jovyan/.bashrc source ~/.profile

6) DAG(Directed Acyclic Graph, 비순환 방향 그래프)파일을 저장할 디렉토리를 만듭니다

mkdir $AIRFLOW_HOME/dags

기본 폴더 구조는 아래와 같습니다

airflow ├── airflow.cfg <- airflow 환경설정 파일 ├── airflow.db <- 데이터베이스(SQLite)파일 ├── dags <- DAG들을 저장하는 디렉토리 │ └── my_first_dag.py <- DAG 정의 파이썬 파일 ├── logs <- 로그파일을 저장하는 디렉토리

airflow.cfg파일에 dags, logs폴더가 지정되어 있습니다.

dags_folder = /home/jovyan/airflow/dags

base_log_folder = /home/jobyan/airflow/logs

7) 설치된 airflow 버전을 확인합니다

airflow version

8) 등록된 DAG 목록을 조회합니다

airflow list_dags

9) dags 폴더에 실습할 my_first.py 을 만듭니다

cd $AIRFLOW_HOME/dags/ && vim my_first.py

[소스코드]

from airflow.models import DAG from airflow.utils.dates import days_ago from airflow.operators.bash_operator import BashOperator args = {'owner': 'jovyan', 'start_date': days_ago(n=1)} dag = DAG(dag_id='my_first_dag', default_args=args, schedule_interval='@daily') t1 = BashOperator(task_id='print_date', bash_command='date', dag=dag) t2 = BashOperator(task_id='sleep', bash_command='sleep 3', dag=dag) t3 = BashOperator(task_id='print_whoami', bash_command='whoami', dag=dag) t1 >> t2 >> t3

10) 코드를 실행해봅니다(문법 오류 정도를 검증할 수 있습니다)

경축! 아무것도 안하여 에스천사게임즈가 새로운 모습으로 재오픈 하였습니다.
어린이용이며, 설치가 필요없는 브라우저 게임입니다.
https://s1004games.com

cd $AIRFLOW_HOME/dags/ && python my_first.py

11) 다시 등록된 DAG 목록을 조회합니다

airflow list_dags

my_first_dag 이름을 가진 DAG를 볼 수 있습니다

12) 새로만든 DAG의 Task를 조회합니다

airflow list_tasks my_first_dag

print_date, print_whoami, sleep 3가지 Task가 있습니다

명령어 두에 --tree를 붙이면 트리형태로 확인할 수 있습니다

13) 테스트로 Task 1개만 실행해봅니다

# [사용법] airflow test dag_id task_id execution_date airflow test my_first_dag print_date 2019-06-01T09:00:00

테스트 목적으로 Task 1개만 단독으로 실행되었습니다

14) airflow 스케쥴러를 실행합니다

airflow scheduler &

15) 스케쥴러를 컨트롤하고 관리할 웹서버를 실행합니다

airflow webserver -p 8882 &

16) 웹브라우저로 접속합니다

http://localhost:8882/admin/

17) DAGs 탭에서 만들었던 my_first_dag을 Off에서 On으로 활성화 합니다

활성화 후 우측의 Links에서 3번째 Graph View를 클릭합니다

18) Graph View를 통해 각 Operator의 순서와 상태를 확인할 수 있습니다

종료되면 진한 녹샌으로 변합니다. 더 자세한 로그를 확인하고자 Operator를 클릭합니다

19) Operator 팝업에서 "View Log"를 클릭합니다

20) 해당 Operator의 실행 차수별 로그를 확인할 수 있습니다

정상적으로 BashOperator인 print_date가 실행되었습니다

21) Gantt 차트를 확인해보겠습니다

print_date -> sleep -> print_whoami 순서로 진행되었고, sleep이 가장 오래 시간이 걸렸습니다

Airflow는 다양한 종류의 Operator를 지원합니다.

위 기초 실습에서는 이해를 돕기 위해 BashOperator 만을 사용했습니다.

끝.

[출처] https://blog.naver.com/wideeyed/221565240108

본 웹사이트는 광고를 포함하고 있습니다.
광고 클릭에서 발생하는 수익금은 모두 웹사이트 서버의 유지 및 관리, 그리고 기술 콘텐츠 향상을 위해 쓰여집니다.

이 게시물을

번호	제목	글쓴이	날짜	조회 수
15	[python 데이터분석] [jupyter] 주피터 노트북에 이미지 삽입	졸리운_곰	2025.09.06	526
14	[python 데이터분석] [Python] Streamlit 사용법 (python 데이터분석 웹 만들기)	졸리운_곰	2024.12.22	462
13	[python 데이터분석] Anaconda : Error while loading conda entry point: conda-libmamba-solver (libarchive.so.19: cannot open shared object file: No such file or directory)	졸리운_곰	2024.12.14	520
12	[python 데이터분석] Anaconda \| Conda update 반영 안됨(update 후에도 버전 변경 없음)	졸리운_곰	2024.11.18	431
11	[python 데이터분석] Keeping Anaconda Up To Date	졸리운_곰	2024.05.30	552
10	[python 데이터 분석] 국내 경제 100대 통계지표	졸리운_곰	2024.02.18	672
9	[python 데이터 분석] Python 에서 R언어 패키지 호출 : Calling R From Python With rpy2	졸리운_곰	2024.01.28	676
8	[python 데이터 분석] 파이썬을 활용한 코스피, 달러 환율정보 수집부터 차트 시각화까지	졸리운_곰	2023.12.11	735
7	[Python 데이터분석][pandas] [Python pandas] DataFrame의 문자열 칼럼을 숫자형으로 바꾸기 : pd.to_numeric(), DataFrame.astype()	졸리운_곰	2023.12.09	357
6	[Python 데이터분석] [Python 환경설정] VS code 설치 및 Anaconda와 연동하기	졸리운_곰	2023.03.17	527
5	[Python 데이터분석][python 데이터분석 프로덕션] [Python] Docker를 사용한 Dash 웹앱 생성	졸리운_곰	2021.12.10	403
4	[Python 데이터분석] [pandas] 공공데이터(csv) 활용시 한글 깨짐 현상 해결	졸리운_곰	2021.09.30	592
3	[Python 데이터분석] 공공데이터포털::공휴일 데이터 조회 (REST API)	졸리운_곰	2021.09.30	358
2	[Python 데이터 분석] pandas의 to_csv()를 사용해서 csv 파일로 저장하기(save 하기)	졸리운_곰	2021.09.29	603
1	[Python 데이터 분석] 데이터 과학을 단순하게 만드는 3가지 Python 패키지	졸리운_곰	2021.09.24	513

첫 페이지 1 끝 페이지

쓰기

태그

[python] Apache Airflow 소개 및 실습하기(기초)

[python] Apache Airflow 소개 및 실습하기(기초)

경축! 아무것도 안하여 에스천사게임즈가 새로운 모습으로 재오픈 하였습니다. 어린이용이며, 설치가 필요없는 브라우저 게임입니다. https://s1004games.com

댓글 0

로그인

경축! 아무것도 안하여 에스천사게임즈가 새로운 모습으로 재오픈 하였습니다.
어린이용이며, 설치가 필요없는 브라우저 게임입니다.
https://s1004games.com