airflow

초기설치

airflow는 python기반입니다.

python설치후 아래 명령으로 설치 및 초기화

pip install apache-airflow
airflow db init

위는 sqlite기반으로 설치되며 병렬처리에 한계가 있어서 postgres에 설치하는 방법이 따로 있음(아래 더보기 참조)

1. postgres에 설치하기 위해서는 airflow를 설치할때 아래 명령사용

pip install 'apache-airflow[postgres]'

2. PostgreSQL 설치 및 설정: PostgreSQL이 설치되어 있지 않다면 먼저 설치해야 합니다. 
Airflow에서 사용할 데이터베이스와 사용자를 생성해야 합니다.
PostgreSQL에서 다음과 같은 명령어를 실행하여 데이터베이스와 사용자를 생성할 수 있습니다.

CREATE DATABASE airflow_db;
CREATE USER airflow_user WITH PASSWORD 'your_password';
GRANT ALL PRIVILEGES ON DATABASE airflow_db TO airflow_user;

3. airflow.cfg 수정: 이제 airflow.cfg 파일을 열어서 sql_alchemy_conn 설정을 수정 합니다.

sql_alchemy_conn = postgresql+psycopg2://airflow_user:your_password@localhost/airflow_db

다음 커멘드를 실행하여 웹서버및 스케줄러 띄움

airflow webserver -p 8080
airflow scheduler

단 위의 각 라인이 실행후 블럭되는 형태라, supervisorctl등으로 데몬으로 띄우는게 좋음(아래 더보기 참조)

1. Supervisor 설치: 먼저 Supervisor를 설치해야 합니다. Ubuntu에서는 다음 명령어로 설치할 수 있습니다.

sudo apt-get install supervisor

2. Supervisor 설정 파일 작성: Supervisor는 설정 파일을 사용하여 관리할 프로세스를 정의합니다. 각 프로세스는 설정 파일의 [program] 섹션에 정의됩니다. Airflow 웹 서버와 스케줄러를 위한 설정 파일을 작성해야 합니다. 예를 들어, /etc/supervisor/conf.d/airflow-webserver.conf 파일을 다음과 같이 작성할 수 있습니다.

[program:airflow-webserver]
command=/home/sevity/.local/bin/airflow webserver -p 8081
user=sevity
autostart=true
autorestart=true
redirect_stderr=true
environment=PATH="/home/sevity/.local/bin:%(ENV_PATH)s"

3. 이와 유사하게, /etc/supervisor/conf.d/airflow-scheduler.conf 파일을 다음과 같이 작성할 수 있습니다.

[program:airflow-scheduler]
command=bash -c "trap '/path/to/kill_gunicorn.sh' EXIT; /home/sevity/.local/bin/airflow scheduler"
user=sevity
autostart=true
autorestart=true
redirect_stderr=true
environment=PATH="/home/sevity/.local/bin:%(ENV_PATH)s"

kill_gunicorn.sh의 내용은 아래와 같으며, airflow를 죽여도 gunicorn이 남는 것을 같이 종료해주는 코드다.

#!/bin/bash
PIDS=$(pgrep -f gunicorn)
for PID in $PIDS
do
    PARENT_PID=$(ps -o ppid= -p $PID)
    kill -9 $PID $PARENT_PID
done

 

 

http://server_ip:8080으로 접속하면 로그인 창이 뜨는데, 로그인을 위해 리눅스 콘솔에서 아래명령어 수행

airflow users create \
    --username your_username \
    --firstname your_firstname \
    --lastname your_lastname \
    --role Admin \
    --email your_email@example.com

 

dag파일 샘플

sevity@sevityubuntu:~$ cat ~/airflow/dags/deploy_online_judge_services.py
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
from pendulum import timezone

KST = timezone("Asia/Seoul")

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 7, 28, tzinfo=KST),
    'email': ['your-email@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'build_online-judge_services', default_args=default_args, schedule_interval="0 22 * * *")

# 이전에 stop 명령 실행
t0 = BashOperator(
    task_id='stop_services',
    bash_command='sudo supervisorctl stop online-judge_auth-service online-judge_frontend-service',
    dag=dag
)

t1 = BashOperator(
    task_id='build_auth_service',
    bash_command='cd ~/workspace/online_judge/auth-service && ./mvnw clean install && docker build -t auth-service .',
    dag=dag)

t2 = BashOperator(
    task_id='build_frontend_service',
    bash_command="""
        export NVM_DIR="/home/sevity/.nvm"
        [ -s "$NVM_DIR/nvm.sh" ] && \. "$NVM_DIR/nvm.sh"
        nvm use 20.5.0
        cd ~/workspace/online_judge/frontend-service && npm install && npm run build
        docker build -t frontend-service .
    """,
    dag=dag
)

# 이후에 start 명령 실행
t3 = BashOperator(
    task_id='start_services',
    bash_command='sudo supervisorctl start online-judge_auth-service online-judge_frontend-service',
    dag=dag
)


t0 >> t1 >> t3
t0 >> t2 >> t3

마지막에 dag에서 선후관계를 설정해주는건데, 단순 선형관계 말고 아래처럼 다이아몬드처럼 할때는 위처럼 하면된다.

중간에 실패한 task가 있을때 그거 부터 시작하려고 하면 clear라는 명령을 주면 된다.

Future옵션을 끄면 후속작업을 안하게 할 수도 있다.

 

트러블슈팅

 

supervisor와 연동하기

airflow webser는 airflow프로세스를 죽여도 gunicorn관련 프로세스가 남고, 

airflow scheduler는 executor, worker 등이 남는다. 관련해서 아래처럼 kill script를 만들고 연동해줬다.

# kill_webserver.sh
#!/bin/bash

# Gunicorn 프로세스 종료
PIDS=$(pgrep -f gunicorn)
for PID in $PIDS
do
    PARENT_PID=$(ps -o ppid= -p $PID)
    kill -9 $PID $PARENT_PID
done


# kill_scheduler.sh
#!/bin/bash

# airflow scheduler 프로세스 종료
AIRFLOW_SCHEDULER_PIDS=$(pgrep -f "airflow scheduler")
for PID in $AIRFLOW_SCHEDULER_PIDS
do
    PARENT_PID=$(ps -o ppid= -p $PID)
    kill -9 $PID $PARENT_PID
done

# airflow executor 프로세스 종료
PIDS=$(pgrep -f "airflow executor")
for PID in $PIDS
do
    PARENT_PID=$(ps -o ppid= -p $PID)
    kill -9 $PID $PARENT_PID
done

# airflow worker 프로세스 종료
PIDS=$(pgrep -f "airflow worker")
for PID in $PIDS
do
    PARENT_PID=$(ps -o ppid= -p $PID)
    kill -9 $PID $PARENT_PID
done

supervisor conf.d에 다음처럼 연동해준다.

# cat /etc/supervisor/conf.d/airflow-webserver.conf
[program:airflow-webserver]
command=bash -c "trap '/home/sevity/airflow/kill_webserver.sh' EXIT; /home/sevity/.local/bin/airflow webserver -p 8081"
user=sevity
stopasgroup=true
killasgroup=true
autostart=true
autorestart=true
redirect_stderr=true
environment=PATH="/home/sevity/.local/bin:%(ENV_PATH)s"


# cat /etc/supervisor/conf.d/airflow-scheduler.conf
[program:airflow-scheduler]
command=bash -c "trap '/home/sevity/airflow/kill_scheduler.sh' EXIT; /home/sevity/.local/bin/airflow scheduler"
user=sevity
stopasgroup=true
killasgroup=true
autostart=true
autorestart=true
redirect_stderr=true
environment=PATH="/home/sevity/.local/bin:%(ENV_PATH)s"