이제 로컬에 Airflow를 설치해보자!

python venv 설정

venv(Virtual Environment)는 일종의 샌드박스이다.
venv 안에서 설치한 파이썬 패키지는 외부에 영향을 받거나 미치지 않고,
venv를 activate 해야만 설치했던 패키지에 접근 할 수 있다.
airflow를 설치하기 전에, venv를 이용해서 외부 영향이 없도록 하자.

$ python3 -m venv {venv_name} #venv 생성
$ source {venv_name}/bin/activate #venv 활성화
(venv) $ deactivate #venv 비활성화

venv과 활성화되면 terminal 앞쪽에 venv 이름이 괄호로 묶여 나오게 된다.

Airflow 설치

wheel 설치

Airflow에 이용될 파이썬 모듈인 wheel을 설치하자.

$ pip install wheel

airflow 설치

$ AIRFLOW_VERSION=2.2.3 # 최신 버전
$ PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)"
$ CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
$ pip install "apache-airflow[async,postgres,google]==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"

이렇게만 설치는 끝났다!

airflow 초기 설정

Metastore의 초기화와 Airflow 실행에 필요한 파일들(SQLite DB file 등)을 초기 생성해주는 명령어이다.

$ airflow db init

파일 구조

• airflow.cfg : airflow configuration file
• airflow.db : SQLite DB file
• /logs : Log directory for airflow(Task, Scheduler, etc..)
• unittests.cfg : 실제 airflow에 영향을 미치지 않고 configuration setting을 테스트하기 위해 존재하는 파일
• webserver_config.py : 웹서버 설정을 위한 config 파일

웹서버 실행하기

$ airflow webserver --port 8080

이후 http://localhost:8080 으로 접속하면, 웹서버에 접근할 수 있다!
하지만 접근 이후 계정을 생성한 적이 없어서 로그인이 불가능하다!
이럴때는 Airflow CLI를 이용하여, 계정을 생성해줘야한다.

유용한 CLI 명령어

users create

$ airflow users create \
    --username admin \
    --firstname Peter \
    --lastname Parker \
    --role Admin \
    --email spiderman@superhero.org

db init

$ airflow db init

db upgrade

airflow의 버전을 업데이트하고 싶을 때 이용한다

$ airflow db upgrade

db reset

metastore의 데이터를 전부 지우는 명령어임으로, 운영환경에서는 사용하지 않도록 주의해야한다.

$ airflow db reset

webserver

$ airflow webserver

scheduler

$ airflow scheduler 

dags list

$ airflow dags list

tasks list

$ airflow tasks list {dag_id} 

trigger dag

$ airflow dags trigger -e 2022-01-01 example_xcom_args

1.10.X 버전 명령어

위에 언급한 명령어들은 모두 2.X 버전의 명령어들이다.
혹시 1.10.X 버전을 이용하는 사용자들은 아래 명령어를 참고하는게 좋을 것 같다.

initdb

$ airflow initdb

web server

$ airflow webserver --port 8080

list dags

$ airflow list_dags

list tasks of the dag_id

$ airflow list_tasks {dag_id}
$ airflow list_tasks hello_world

test specific tasks

$ airflow test {dag_id} {task_id} {execution_date}