[Airflow] 로컬에 Airflow 설치 및 시작하기

1️⃣ Install Airflow

pyenv 가상환경 사용

pyenv virtualenv 3.9.15 airflow-env

cd ~~
pyenv activate airflow-env

# Airflow needs a home. `~/airflow` is the default, but you can put it
# somewhere else if you prefer (optional)
export AIRFLOW_HOME=~/airflow

# Install Airflow using the constraints file
AIRFLOW_VERSION=2.4.2
PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)"
# For example: 3.7
CONSTRAINT_URL="<https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt>"
# For example: <https://raw.githubusercontent.com/apache/airflow/constraints-2.4.2/constraints-3.7.txt>
pip3 install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"

# The Standalone command will initialise the database, make a user,
# and start all components for you.
airflow standalone

# Visit localhost:8080 in the browser and use the admin account details
# shown on the terminal to login.
# Enable the example_bash_operator dag in the home page

Airflow is ready
standalone | Login with username: admin  password: VxfhNC5kkGewzfZf

2️⃣ Set up MySQL

해당 MySQL은 Airflow의 메타 데이터를 저장하기 위한 DB.

기본 DB는 SQLite → 연습용이나 소규모 프로젝트에는 기본 DB 써도 괜찮음.

MySQL | Mac에 MySQL 설치하기 (M1칩)

create user 'airflow'@'%' identified by 'airflow1234';
CREATE Database airflowdb;
grant all privileges on airflowdb.* to 'airflow'@'%' with grant option;
flush privileges;

• Port 확인법
  • SHOW GLOBAL VARIABLES LIKE 'PORT';
• vi ~/airflow/airflow.cfg에 MySQL 경로 설정
• sql_alchemy_conn = mysql://root:root@localhost:3306/airflow

3️⃣ Start Airflow

 

DAG 관리 디렉토리 설정

vi ~/airflow/airflow.cfg

dags_folder = /Users/myksh0903/airflow/dags

디렉토리 안에 DAG 작성

• Sample DAG

from airflow.models import DAG
from airflow.utils.dates import days_ago
from airflow.operators.bash_operator import BashOperator

args = {'owner': 'airflow', '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

Airflow scheduler 실행

DAG 실행하기 위해 scheduler를 먼저 실행해야 한다. scheduler가 dag folder를 바라보고 있어 dag의 내용을 불러온다.

scheduler를 실행해보면 scheduler도 webserver로 구성되어 있다. but backend 부분만 구성되어 있다.

airflow scheduler &

scheduler를 컨트롤하고 관리할 웹서버를 실행

웹 UI로 관리하기 위해 웹 서버를 실행시킨다.

airflow webserver -p 8080 &

** Airflow는 webserver, scheduler, executor가 각각 구성되어 있다.