Airbnb에서 만든 Workflow 관리 오픈소스 소프트웨어입니다.
많은 기업에서 데이터 파이프라인 운용에 인기 있게 사용되고 있습니다.
DAGs(Directed Acyclic Graphs) 형태로 Workflow 수행합니다.

사전 준비사항

• MySQL 설치
• Redis 설치

Install

# Airflow 설치
pip3 install apache-airflow

# MySQL 플러그인 설치
pip3 install 'apache-airflow[mysql]'

# Celery 설치
pip3 install celery

Airflow와 메타데이터를 저장할 MySQL 플러그인을 설치합니다.

CeleryExecutor 사용을 위해서 Celery를 설치합니다.

참고로 SQLite가 디폴트로 사용이 가능합니다. Executor는 Sequential Executor만 사용 가능합니다. 다른 Executor 사용을 위해서는 메타데이터를 저장할 별도의 데이터베이스 구성을 해야 합니다.

MySQL 환경 준비

mysql -u root -p

-- 스키마 생성
create database airflow;

-- 유저 생성
create user 'airflow'@'%' identified by 'Airflow!1';

-- airflow 유저에 airflow 스키마 권한 생성
grant all privileges on airflow.* to 'airflow'@'%';
flush privileges;

Root 계정으로 로그인하여 Schema, User를 생성합니다.

Airflow 설정

# Timezone 설정
default_timezone = Asia/Seoul

# 메타데이터 저장소 설정
sql_alchemy_conn = mysql://airflow:Airflow!1@localhost:3306/airflow

# Executor DB 설정
result_backend = db+mysql://airflow:Airflow!1@localhost:3306/airflow

# Redis 설정
broker_url = redis://127.0.0.1:6379/0

# DAG 리스트 갱신 설정 (테스트니까 10초로)
dag_dir_list_interval = 10

# Executor를 설정
executor = CeleryExecutor

airflow.cfg를 수정합니다.

Airflow 명령어

# Airflow DB 초기화
airflow db init

# Airflow Admin 권한의 유저 생성
airflow users create --role Admin --username [USERNAME] --email admin --firstname [FIRST_NAME] --lastname [LAST_NAME] --password [PASSWORD]

# Airflow Webserver 기동
# -p : 포트 설정
# -D : 데몬 설정
airflow webserver -p 8080 -D

# Airflow Scheduler
# -D : 데몬 설정
airflow scheduler -D

# Celery Worker
# -q : Queue 이름
airflow celery worker -q [QUEUE_NAME]

Celery Worker는 airflow.cfg에 명시하여 자동으로 worker가 기동이 됩니다.
만약 별도의 Worker를 띄우고자 한다면 CLI로 운용할 수 있습니다.

Celery

pip install로 별도로 Celery를 설치하는 것을 보고 뭐지? 하셨을 수 있을 거 같아요.
Celery는 Python으로 작성된 Asynchronous task queue/Job Queue입니다.
Job을 Broker를 통해서 전달하면 하나 이상의 Worker에서 처리하는 구조입니다.

Airflow Celery Executor 사용을 위해서 본 문서에서는 Broker로 Redis를 사용하였습니다.

참조문서

• Airflow Celery Executor
• Celery를 이용한 긴 작업 처리