[Airflow] 시작

식빵·2025년 5월 31일

airflow data-pipline etl python

Airflow

목록 보기

1/9

🤔 Airflow ?

Apache Airflow는 오픈소스 배치 워크플로우 오케스트레이션 플랫폼입니다.
Airflow 의 특징은 다음과 같습니다.

워크플로우를 코드로 정의:
모든 워크플로우는 Python 코드로 작성되며,
이를 통해 동적 생성, 확장 및 유연한 파라미터 설정이 가능합니다.
강력한 스케줄링 기능:
DAG(Directed Acyclic Graph)를 활용하여 작업 간의 의존성을 설정하고,
정해진 주기에 따라 실행할 수 있습니다.
확장성:
다양한 기술과 연결할 수 있는 연산자(Operators)를 제공하며,
필요에 따라 새로운 연산자를 추가할 수도 있습니다.
웹 인터페이스 제공:
워크플로우의 실행 상태를 시각적으로 확인하고,
개별 작업의 로그를 조회하거나 재시도할 수 있습니다.

더 궁금하시면 공식 apache airflow 소개글을 확인해보시 바랍니다.

📲 Airflow 로컬 설치

일단 설치를 해야 뭐든 할 수 있겠죠?
로컬에 설치부터 해봅시다.

이 글에서는 Airflow 2.10.5 버전을 사용했습니다.
Airflow 3.0.1 이 나오긴 했는데, 나온지 얼마 안되서 2 버전대를 사용했습니다.

설치환경은 리눅스의 OS 여야만 합니다.
저는 window 사용자여서 WSL2 (Ubuntu 22.04) 환경에서 작업을 했습니다.

Docker 로 설치

airflow 를 Docker 를 사용해서 로컬에 설치해보겠다.
설치 방법은 공식 apache airflow 의 글을 보고 따라하겠습니다.

cd ~
mkdir airflow-workspace && cd airflow-workspace

# Fetching docker-compose.yaml
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.10.5/docker-compose.yaml'

# Initializing Environment
mkdir -p ./dags ./logs ./plugins ./config
echo -e "AIRFLOW_UID=$(id -u)" > .env

# Initialize the database
sudo docker compose up airflow-init

이러고 나서 아래처럼 문구가 나오면 일단 기본적인 준비는 끝난겁니다.

airflow-init_1       | Upgrades done
airflow-init_1       | Admin user airflow created
airflow-init_1       | 2.10.5
start_airflow-init_1 exited with code 0

이제 airflow 를 실행해봅시다.

sudo docker compose up

이후에 계속해서 뭔가 지지고 볶습니다.
하지만 일단 너무 걱정하지 마시고, 다른 bash 창을 하나 키고,
docker ps 를 통해서 현재 컨테이너 상태를 확인합니다.
(아래그림 참고, 클릭하면 크게 볼 수 있습니다)

여기서 봐야할 건 새로 생긴 컨테이너의 목록을 먼저 확인해야 합니다.

airflow-scheduler
airflow-worker
airflow-webserver
airflow-trigger
postgres
redis

이렇게 6개가 보인다면 일단 갯수는 맞는 겁니다.
(아, 그런데 처음에는 airflow-init 도 잠깐 보일 수 있습니다!)

다음으로 봐야할 건 컨테이너의 STATUS 입니다.
STATUS 가 (healty) 로 표시된 상태여야 airflow 를 정상적으로 사용할 수 있는 겁니다.

참고로 docker 를 껏다가 다시 켜도 자동으로 저 모든 컨테이너들이 다시 기동됩니다.
이때도 마찬가지로 STATUS 가 healty 상태가 될 때까지 기다리시면 됩니다.

airflow ui 접속

이제 설치가 잘됐는지 최종적인 검사를 해봅시다.
브라우저로 http://localhost:8080 를 입력해서 airflow ui 에 접속합니다.

위처럼 로그인 화면이 나오는데,
Username, Password 모두 airflow 를 입력하고 로그인 합니다.

위처럼 이쁜 화면이 나오네요.
이렇게 해서 설치는 끝입니다.

📌 기본 용어 익히기

이제는 Airflow 를 사용하기 위해서 기본적으로 알아야할 용어들을 알아보겠습니다.

DAG

DAG(Directed Acyclic Graph) 는 Airflow 의 핵심 구성요소로,
여러 작업(Task)을 모아 의존성과 관계를 구성하여 실행 순서를 정의하는 역할을 합니다.

즉 DAG 자체가 곧 하나의 워크플로우라고 생각하면 됩니다.

또한 DAG 는 순환되지 않는(=Directed Acyclic) 워크플로우를 의미하기도 합니다.
순환이 되는 방식으로 Task 의존성을 구성하면 에러가 납니다.

Operator/Task

DAG 는 여러 Task 를 모아서 만들어진 워크플로우라고 했습니다.
즉 Task 가 실제로 DAG 내에서 발생하는 연산인 것이죠.

추후에 DAG 를 python 으로 코드를 작성해서 반영하게 됩니다.
다만 이때는 저희는 Task 가 아닌 Operator 라는 것을 사용해서 DAG 를 구성합니다.
(아래 그림 참고)

이렇게 작성된 python 코드는 추후 airflow 에 의해 parsing 되어서
DAG 가 생성되고, airflow 가 이 DAG 를 trigger(=실행)하면
작성한 Operator 가 인스턴스화 되어서 Task 가 되고 그게 동작을 합니다.

쉽게 얘기해서 Operator 는 클래스이고, Task 는 인스턴스입니다.

🖥️ 개발환경 구성

이제 Airflow 에 DAG 를 저희가 직접 구성할 건데,
먼저 개발환경을 세팅해야 합니다. 그 방법부터 알아보죠.

파이썬 다운로드

airflow 의 DAG 는 결국 python 코드 덩어리입니다.
그러니 저희도 python 을 사용해서 DAG 를 작성해야겠죠?

먼저 airflow 가 사용하는 python 버전과 동일한 python 을 설치해야됩니다.
먼저 airflow 의 python 버전을 확인해보겠습니다.

# worker 컨테이너의 python 버전 화인
docker exec $(docker ps --format '{{.Names}}' | grep worker) python -V

저는 위처럼 입력하니 Python 3.12.9 가 나오네요.
저도 Host PC 에 Python 3.12.9 를 설치하겠습니다.

다만 저는 추후에 다양한 python 버전들을 유연하게 변경하여 사용하기 위해서
pyenv 라는 프로그램을 통해서 python 3.12.9 를 설치하겠습니다.

# pyenv 의존 라이브러리 설치
sudo apt install build-essential libssl-dev zlib1g-dev \
libbz2-dev libreadline-dev libsqlite3-dev curl \
libncursesw5-dev xz-utils tk-dev libxml2-dev libxmlsec1-dev libffi-dev liblzma-dev

curl -fsSL https://pyenv.run | bash 

echo 'export PYENV_ROOT="$HOME/.pyenv"' >> ~/.bashrc
echo '[[ -d $PYENV_ROOT/bin ]] && export PATH="$PYENV_ROOT/bin:$PATH"' >> ~/.bashrc
echo 'eval "$(pyenv init - bash)"' >> ~/.bashrc

# 이후에 쉘을 다시 실행합니다. (환경변수 세팅 때문)

pyenv install --list | grep ' 3.12.'

만약 위처럼 3.12.9 가 있다면 바로 설치하면 되고,
그렇지 않다면 pyenv update 라고 먼저 입력해서 업데이트 받고 설치해주세요.
(제법 시간이 걸립니다. 약간의 인내심을 가지시길 바랍니다 😅)

# 파이썬 3.12.9 설치
pyenv install 3.12.9

# pyenv 는 여러 파이썬 버전이 있을 수 있는데, 
# default 로 3.12.9 를 사용
pyenv global 3.12.9

# 파이썬 버전 확인
python -V

venv 환경 구성 + library 설치

파이썬 코딩을 위한 디렉토리를 하나 만들어주고,
그 디렉토리에 venv 환경을 구성하고 pip install 을 통해
airflow 라이브러리를 설치해보겠습니다.

# 디렉토리 생성
mkdir ~/python-workspace && cd ~/python-workspace

# venv 환경 구성
python -m venv .venv

# venv 활성화
chmod u+x .venv/bin/activate
source .venv/bin/activate

# airflow 라이브러리 설치
pip install "apache-airflow[celery]==2.10.5" \
	--constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.5/constraints-3.8.txt"

참고로 저 pip install 명령어는 공식 사이트에서 복사해온 겁니다.

visual studio code 환경 구성

visual studio code 로 해당 디렉토리를 열고 개발을 시작하면 됩니다.
다만 저처럼 WSL 환경에서 작업하시는 분들은 아래와 같이 조금 더 작업을 해줘야 합니다.

위 그림처럼 Remote Development 이라는 플러그인을 검색해서 설치해주세요.

좌측에 Remote Explorer 라는 아이콘을 클릭하고,
자신이 WSL 명칭 옆에 Connect in New Window 라는 아이콘을 클릭해주세요.

위처럼 새 창으로 뜨면, 그림처럼 숫자가 적힌대로 클릭하고,
마지막 3번에서 아까 만든 ~/python-workspace 디렉토리를 선택하면 됩니다.
(이미지를 클릭하면 크게 볼 수 있습니다!)

추가로 Python 플러그인을 설치해줍니다.

그리고 나서 ctrl + shift + p 를 입력한 후
Python: Select Interpreter 를 검색하고 Enter!

위 그림처럼 .venv/bin/python 이라는 경로가 보이는 인터프리터를 선택해주세요.

마지막으로 .vscode 라는 디렉토리를 하나 생성하고,
그 아래에 settings.json 을 생성해줍니다.
그리고 내용을 다음과 같이 작성합니다.

{
    "python.analysis.autoImportCompletions": true,
    "python.analysis.packageIndexDepths": [
        {
            "name": "",
            "depth": 15,
            "includeAllSymbols": false
        }
    ]
}

이후에 한번 vscode 를 껏다가 다시 켜줍니다.
(가끔 세팅이 잘 안될 때가 있어서 이 작업을 해줍니다)

여기까지가 WSL 환경의 python 개발환경 세팅입니다.

🐍 DAG 생성 및 적용

자 이제 개발환경 구성이 끝났으니 DAG 를 만들고 airflow 에 넘겨줘 봅시다.
먼저 아래와 같이 코딩해줍니다.

# 파일명 : my_hello_world_dag.py
from airflow import DAG
import datetime
from airflow.operators.bash import BashOperator
import pendulum

with DAG(
    dag_id="my_hello_world_dag",  # 파일명과 일치시키는 것이 관례
    schedule="0 0 * * *",
    start_date=pendulum.datetime(2025, 5, 31, tz="Asia/Seoul"),
    catchup=False,
    dagrun_timeout=datetime.timedelta(minutes=60), # timeout 60분
    tags=["hello world", "sample"], # 태그, 검색시 좋음
    # params={"example_key": "example_value"}, # Task 들의 공통적으로 사용될 파라미터 작성
) as dag:

	# BashOperator = Bash 명령어를 실행시키는 Operator 
    bash_t1 = BashOperator(
        task_id="bash_task1", 
        bash_command="echo hello",
    )

    bash_t2 = BashOperator(
        task_id="bash_task2",  
        bash_command="echo world",
    )

    bash_t1 >> bash_t2  # type: ignore

DAG 생성자에 사용되는 key 들의 의미는 다음과 같습니다.

dag_id : DAG 를 식별하는 ID 이며, airflow ui 에서는 DAG 목록의 제목으로 보입니다.

schedule : cron 문법을 통한 스케줄링입니다.

start_date : 시작 시간을 의미합니다. tz='Asia/Seoul' 을 꼭 지정해주세요!

catchup:
만약 현재 일자가 2025.06.02 이고, startup_date 가 2025.05.02 라고 가정해봅시다.
그리고 catchup=True 로 세팅하면, 2025.05.02 부터 현재일자까지
한번에 Task 들이 동시 다발적으로 실행됩니다. (일자별 순서대로 동작 X)
말그대로 여태까지 못했던 일들을 catchup 한다는 의미입니다.

dagrun_timeout : timeout 시간입니다.

tags : airflow ui 에 보면 DAG 목록 밑에 태그명이 보이는데, 이런 태그들을 통해서
검색시에 이점을 갖을 수 있습니다.

params : Operator 들에게 공통적으로 사용할 파라미터를 작성한 겁니다.

이제 이 python 코드 파일을
airflow 의 docker compose up 명령어 입력한 곳에 있는 dags 디렉토리로 옮겨주면 끝입니다.

cp my_hello_world_dag.py ~/airflow-workspace/dags

이후에 airflow ui 를 엽니다. Auto-refresh 를 활성화하고
한 1~2분 정도 지나면 저희가 만든 DAG, my_hello_world_dag 화면에 보이기 시작합니다.
(화면에 보이기까지 제법 시간이 걸립니다!)

저희가 지정한 start_date 까지 기다리기 귀찮으니,
화면 우측에 있는 Trigger DAG 버튼을 클릭해서 강제로 시작합니다.

위 그림처럼 순서대로 클릭해주세요.

로그를 통해서 BashOperator 가 정상적으로 명령어를 수행한 것을 확인할 수 있습니다.
이로써 정상적인 구동을 확인했습니다.

확인이 끝났으니 이제 DAG 를 PAUSE 시켜줍니다.

여기까지가 제가 아는 가장 기본 준비 과정입니다.
그럼 이만 글을 마치도록 하겠습니다.

✨ 참고 링크

식빵

백엔드 개발자로 일하고 있는 식빵(🍞)입니다.

다음 포스트