Apache Airflow Research - mozzihozzi/DevOps GitHub Wiki

Apache Airflow

Apache Airflow 조사

소개

설치방법

사용가이드

커맨드 정리

1. 소개

1.1. Apache airflow 란?

enter image description here https://airflow.apache.org/

오픈소스 워크플로우 스케줄링, 모니터링 플랫폼
Airbnb에서 개발
주로 빅데이터에서 데이터 프로세스를 관리하기 위해 사용함
Python 언어로 사용

1.2. 특징

1. 복잡한 프로세스를 flow diagram 형태로 확인 가능
2. Python으로 task를 작성 및 관리 + 이쁘고 편리한 UI
3. 각 task의 실행 시간, 이력을 쉽게 확인 가능
4. 필요한 경우 특정 task만 실행 가능
5. 각 task를 병렬로 실행 가능
6. 독립된 자체 스케쥴러로 각 사용자가 독립적으로 수행 가능

종합하면, 사용자가 workflow를 쉽게 관리 및 모니터링 할 수 있음

1.3. dag(Directed Acyclic Graph, 방향성 비순환 그래프)

workflow를 구성하는 task들의 집합들을 dag라고 함
dag는 한쪽 방향으로만 진행되고 순환되지 않는 그래프를 의미함
여러 task들이 순서와 종속성을 가짐
dag안에서 task들은 Operator로 부터 정의하고 Operator들은 여러 종류들이 있음

1.4. 기본 폴더 구조

설치된 airflow 디렉터리의 airflow.cfg, airflow.db를 통해 설정을 변경하거나 dag, log들이 저장된다.

airflow
├── airflow.cfg
├── airflow.db
├── dags
│   ├── dags1.py
│   └── dags2.py
│   └── ...
├── logs
├── ...

2. 설치 방법

2.1. Prerequisites

python
pip
venv(optional)

2.2. 설치 과정

apache-airflow 설치, default로 ~/airflow 경로에 설치함 설치 경로를 바꾸고 싶으면 export AIRFLOW_HOME=~/{path} 으로 수정 후 설치 gcp(google cloud platform) or postgres에서 설치할 경우에는 pip install apache-airflow[postgres, gcp]를 사용

pip install apache-airflow

conda로 설치할 경우 아래의 command를 사용

conda install -c conda-forge airflow

airflow 버전 확인

airflow version

airflow에서 사용할 DB 초기화 airflow.cfg에서 DB설정을 하지 않으면 default로 SQLite 사용

airflow initdb

UI를 실행할 웹서버 실행, default 포트는 8080, url은 http://localhost:{port}

airflow webserver -p {port}

airflow 스케쥴러 실행

airflow scheduler

3. 사용 가이드

Airflow Documentation
- https://airflow.apache.org/docs/stable/ui.html#task-instance-context-menu

3.1. dag 작성

airflow는 Operator를 통해 task를 정의함. Operator는 다양한 종류가 있음.

DummyOperator
BashOperator
PythonOperator
Dining Operators
Google Cloud Operators
Papermill

3.1.1. PythonOperator

PythonOperator parameters

task_id : 각 task를 구분하기 위한 task id, 이름은 Unique해야 함
python_calllable :실제 호출된 python 함수 이름
provide_context : python 함수 호출 시 해당 함수에서 사용될 수 있는 기본적인 argument 값을 넘겨줄 지 여부
op_kwargs : 기본 argument 외에 추가로 넘겨줄 parameter 정의
dag : default dag 이름, 보통 dag를 써줌

dag 작성 예시

    from airflow import DAG
    from airflow.operators.dummy_operatorimport DummyOperator
    from airflow.operators.python_operatorimport PythonOperator
	
	def python_task1():
		...
	
	def python_task2():
		...	    

    dag= DAG(‘{dag_filename}', description='Simple tutorial DAG', schedule_interval='0 12 * * *', start_date=datetime(2019, 1, 20), catchup=False)
    
	dummy_op1 = DummyOperator(task_id='dummy_task', retries=3, dag=dag)
	
	python_op1 = PythonOperator(task_id='python_task1', python_callable={function_name}, dag=dag)
	python_op2 = PythonOperator(task_id='python_task2', python_callable={function_name}, dag=dag)
	
	dummy_op1 >> [hello_op1, hello_op2]