위 버전과 저장소는 오래된 것이라 저장소가 존재하지 않을 수 있다.
Airflow 2.1.2 · MySQL · CeleryExecutor 기록이다. 현행 버전은 Airflow v3.3.2 Celery 설치 가이드 를 볼 것. 3.x 에서는airflow db init→airflow db migrate,airflow webserver→airflow api-server로 바뀌었고airflow dag-processor를 따로 띄워야 한다.airflow users create는 FAB auth manager 를 쓸 때만 있다.
Airbnb 에서 개발한 워크플로우 관리 오픈소스 툴.
airflow celery worker -H worker_name -q queue_name 으로 실행 (따로 celery 설치하느라 고생함)pip download. offline 상태일 때는 이렇게 명시적으로 constraint 를 주지 않으면 오류가 난다.
pip download 'apache-airflow[celery,mysql,redis,crypto]==2.1.2' --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.1.2/constraints-3.8.txt"
pip install
sudo yum install mysql-devel # 없으면 mysql 연결 시 에러남
pip install --no-index -f ./ apache-airflow[celery,mysql,redis,crypto]==2.1.2
확인
airflow version
# 2.1.2
airflow celery -h
# usage: airflow celery [-h] COMMAND ...
#
# Start celery components. Works only when using CeleryExecutor.
#
# positional arguments:
# COMMAND
# flower Start a Celery Flower
# stop Stop the Celery worker gracefully
# worker Start a Celery worker node
my.cnf 에 explicit_defaults_for_timestamp=1.
CREATE DATABASE airflow_celery CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
CREATE USER '${MASKED}' IDENTIFIED BY '${MASKED}';
GRANT ALL PRIVILEGES ON airflow_celery.* TO 'airflow_celery';
ALTER DATABASE `airflow_celery` CHARACTER SET utf8;
user 밑의 airflow 에 폴더가 생긴다.
cd /home/manager/airflow
mkdir logs dags
vi /home/manager/airflow/airflow.cfg
[core]
dags_folder = /home/manager/airflow/dags
# celery 사용하기 위해 설정
executor = CeleryExecutor
# webserver metadata
## pymysql 사용하면, pool 관련 에러가 나옴, 버그라고 함
#sql_alchemy_conn = mysql+pymysql://airflow:airflow@107.107.163.216:14381/airflow?charset=utf8
#sql_alchemy_pool_enabled = FALSE
# pymysql 대신 mysqlclient 사용
# 출처: https://stackoverflow.com/questions/61035608/airflow-scheduler-works-normally-fails-with-d
sql_alchemy_conn = mysql+mysqldb://airflow_celery:airflow_celery@107.107.163.216:14381/airflow_celery
sql_alchemy_pool_enabled = True
[logging]
base_log_folder = /home/manager/airflow/logs
[webserver]
# port 수정
base_url = localhost:14380
web_server_port = 14380
[celery]
# queue, rabbitMq 도 가능
broker_url = redis://107.107.163.214:16379/0
# worker 결과
result_backend = db+mysql://airflow:airflow@107.107.163.216:14381/airflow?charset=utf8
# celery UI
flower_port = 14384
airflow db init
airflow users create --firstname admin --lastname admin --email admin --password admin --username admin --role Admin
airflow webserver -D
airflow scheduler -D
#airflow celery flower -D
airflow celery worker -H worker_bash -q worker_bash -D
airflow celery worker -H worker_celery -q worker_celery -D
pid 를 기준으로 kill 해야 한다. airflow celery stop 은 있지만 airflow stop 은 없다.
kill `cat /home/manager/airflow/airflow-webserver.pid`
kill `cat /home/manager/airflow/airflow-scheduler.pid`
kill `cat /home/manager/airflow/airflow-flower.pid`
kill `cat /home/manager/airflow/airflow-worker.pid`
sqlalchemy.exc.IntegrityError: (MySQLdb._exceptions.IntegrityError) (1364, "Field 'id' doesn't have a default value")
[SQL: INSERT INTO celery_taskmeta (task_id, `status`, result, date_done, traceback) VALUES (%s, %s, %s, %s, %s)]
[parameters: ('7f4a503d-0eeb-4d02-80f6-64b2ac79b697', 'PENDING', None, datetime.datetime(2021, 10, 6, 7, 28, 17, 556083), None)]
(Background on this error at: http://sqlalche.me/e/13/gkpj)