Airflow®를 새 버전으로 업그레이드하기
Airflow®를 새 버전으로 업그레이드하기
새 Airflow 버전에는 데이터베이스 마이그레이션이 포함될 수 있어요. 이 문서는 업그레이드 전 DB 백업, 마이그레이션 실행, MySQL 인코딩 문제 해결 등 업그레이드 전체 과정을 안내해 드려요.
출처: 문서
본문
왜 업그레이드해야 하나요?
새 Airflow 버전에는 데이터베이스 마이그레이션이 포함될 수 있으므로, 업그레이드 대상 Airflow 버전의 스키마 변경사항으로 데이터베이스를 마이그레이션하려면 airflow db migrate를 실행해야 해요. 걱정하지 마세요. 수행할 마이그레이션이 없어도 실행해도 안전해요.
Airflow 버전 x와 y 사이의 변경사항은 무엇인가요?
릴리스 노트에 특정 Airflow 릴리스에 포함된 변경사항이 정리되어 있어요.
업그레이드 준비 - DB 백업
어떤 마이그레이션이든 그 전에 메타데이터 DB를 백업하는 것을 강력히 권장해요. DB에 대한 "핫 백업" 기능이 없다면, 백업이 일관되도록 Airflow 인스턴스를 종료한 후에 해야 해요. 백업을 하지 않고 마이그레이션이 실패하면 절반만 마이그레이션된 상태에 빠질 수 있으며, 백업에서 DB를 복원하고 마이그레이션을 다시 실행하는 것이 유일한 쉬운 해결책일 수 있어요. 예를 들어 마이그레이션 중에 CLI와 데이터베이스 사이의 네트워크 연결이 끊어져 이런 문제가 발생할 수 있으므로, 백업을 하는 것은 이런 문제를 피하기 위한 중요한 예방책이에요.
언제 업그레이드해야 하나요?
virtualenv나 Docker 컨테이너 기반의 커스텀 배포가 있다면, 보통 업그레이드 과정의 일부로 DB 마이그레이션을 수동으로 실행해야 해요.
어떤 경우에는 업그레이드가 자동으로 일어나요. 배포에서 업그레이드가 post-install 액션으로 내장되어 있는지에 따라 달라져요. 예를 들어 post-upgrade hooks가 활성화된 Helm Chart for Apache Airflow를 사용하면, 새 소프트웨어가 설치된 직후 데이터베이스 업그레이드가 자동으로 발생해요. 마찬가지로 모든 Airflow-As-A-Service 솔루션도 UI를 통해 Airflow 업그레이드를 선택하면 자동으로 업그레이드를 수행해요.
업그레이드 방법
Apache Airflow®를 원하는 새 버전을 지정해 다시 설치해요.
부트스트랩된 로컬 인스턴스를 업그레이드하려면, 설치 명령을 다시 실행하기 전에 AIRFLOW_VERSION 환경 변수를 의도한 버전으로 설정할 수 있어요. 패치 버전 단위로 단계적으로 업그레이드하세요: 예를 들어 버전 2.8.2에서 2.8.4로 업그레이드한다면 먼저 2.8.3으로 업그레이드해야 해요. 더 자세한 지침은 Quick Start를 참고하세요.
PyPI 패키지를 업그레이드하려면, 원하는 버전을 constraint로 사용해 환경에서 pip install 명령을 다시 실행해요. 더 자세한 지침은 PyPI에서 설치하기를 참고하세요.
데이터베이스를 수동으로 마이그레이션하려면 환경에서 airflow db migrate 명령을 실행해야 해요. 이 명령은 가상 환경이나 Airflow CLI 명령줄 인터페이스 사용하기와 데이터베이스에 접근할 수 있는 컨테이너에서 실행할 수 있어요.
오프라인 SQL 마이그레이션 스크립트
업그레이드 스크립트를 오프라인으로 실행하고 싶다면, 실행될 SQL 문장을 얻기 위해 -s 또는 --show-sql-only 플래그를 사용할 수 있어요. 시작 Airflow 버전은 --from-version 플래그로, 끝 Airflow 버전은 -n 또는 --to-version 플래그로 지정할 수도 있어요. 이 기능은 Airflow 2.0.0부터 Postgres와 MySQL에서 지원돼요.
Airflow 버전 2.7.0 이상의 사용 예시:
: airflow db migrate -s --from-version "2.4.3" -n "2.7.3"
airflow db migrate --show-sql-only --from-version "2.4.3" --to-version "2.7.3"
Warning
airflow db upgrade는 Airflow 버전 2.7.0부터airflow db migrate로 대체되었으며, 전자는 폐지되었어요.
마이그레이션 문제 처리
MySQL 데이터베이스의 잘못된 인코딩
수동으로 또는 이전 버전의 MySQL로 처음 생성한 오래된 Airflow 1.10 데이터베이스를 사용한다면, 데이터베이스의 원래 문자셋에 따라 새 버전의 Airflow로 마이그레이션하는 데 문제가 생길 수 있고, 마이그레이션이 이상한 오류("key size too big", "missing indexes" 등)로 실패할 수 있어요. 다음 장에서 문제를 수동으로 고치는 방법을 설명해요.
왜 그런 오류가 날까요? MySQL 8 데이터베이스에 권장되는 문자셋/정렬(collation)은 각각 utf8mb4와 utf8mb4_bin이에요. 하지만 이는 MySQL 버전에 따라 달라졌고, 다른 문자셋으로 커스텀 생성된 데이터베이스가 있을 수 있어요. 오래된 버전의 Airflow나 MySQL로 생성된 데이터베이스라면, 데이터베이스 생성 시 인코딩이 잘못되었거나 마이그레이션 중에 깨졌을 수 있어요.
아쉽게도 MySQL은 인덱스 키 크기를 제한하며, utf8mb4를 사용하면 Airflow 인덱스 키 크기가 MySQL이 처리하기에 너무 커질 수 있어요. 그래서 Airflow에서는 모든 "ID" 키가 utf8 문자셋(MySQL 8에서 utf8mb3과 동일)을 사용하도록 강제해요. 이렇게 하면 인덱스 크기가 제한되어 MySQL이 처리할 수 있어요.
마이그레이션을 시도하기 전에 문제를 고치는 단계는 다음과 같아요 (물론 방법을 안다면 자신만의 방식으로 해도 돼요).
Airflow의 내부 데이터베이스 구조는 데이터베이스 ERD 스키마에서, 마이그레이션 목록은 데이터베이스 마이그레이션 참조에서 확인할 수 있어요.
- 실수에 대비해 복원할 수 있도록 데이터베이스를 백업해요.
- 수정해야 할 테이블이 어느 것인지 확인해요. 다음 테이블들을 확인하세요:
SHOW CREATE TABLE task_reschedule;
SHOW CREATE TABLE xcom;
SHOW CREATE TABLE task_fail;
SHOW CREATE TABLE rendered_task_instance_fields;
SHOW CREATE TABLE task_instance;
출력을 반드시 복사해 두세요. 마지막 단계에서 필요해요. dag_id, run_id, task_id, key 컬럼은 다음처럼 utf8 또는 utf8mb3 문자셋이 명시적으로 지정되어 있어야 해요:
``task_id`` varchar(250) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL, # correct
또는
``task_id`` varchar(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin NOT NULL, # correct
필드에 인코딩이 없으면 문제가 돼요:
``task_id`` varchar(250), # wrong !!
아니면 collation만 utf8mb4로 설정된 경우:
``task_id`` varchar(250) COLLATE utf8mb4_unicode_ci DEFAULT NULL, # wrong !!
또는 문자셋과 collation 모두 utf8mb4인 경우:
``task_id`` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, # wrong !!
잘못된 문자셋/collation이 설정된 필드를 고쳐야 해요.
- 수정해야 할 테이블의 외래 키 인덱스를 제거해요 (전부 제거할 필요는 없어요. 수정해야 할 테이블에 대해서만 하면 돼요). 마지막 단계에서 다시 만들어야 하므로 (2단계의
SHOW CREATE TABLE출력이 필요한 이유예요).
ALTER TABLE task_reschedule DROP FOREIGN KEY task_reschedule_ti_fkey;
ALTER TABLE xcom DROP FOREIGN KEY xcom_task_instance_fkey;
ALTER TABLE task_fail DROP FOREIGN KEY task_fail_ti_fkey;
ALTER TABLE rendered_task_instance_fields DROP FOREIGN KEY rtif_ti_fkey;
ID필드를 올바른 문자셋/인코딩으로 수정해요. 인코딩이 잘못된 필드에 대해서만 하세요 (사용할 수 있는 모든 잠재적 명령):
ALTER TABLE task_instance MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE task_reschedule MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE rendered_task_instance_fields MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE rendered_task_instance_fields MODIFY dag_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE task_fail MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE task_fail MODIFY dag_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE sla_miss MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE sla_miss MODIFY dag_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE task_map MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE task_map MODIFY dag_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE task_map MODIFY run_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE xcom MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE xcom MODIFY dag_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE xcom MODIFY run_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
ALTER TABLE xcom MODIFY key VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
- 3단계에서 제거한 외래 키를 다시 생성해요.
제거한 모든 인덱스에 대해 이 과정을 반복해요. Airflow 버전에 따라 인덱스가 약간 다를 수 있지만(예: map_index는 2.3.0에서 추가됨), 2단계에서 준비한 SHOW CREATE TABLE 출력을 유지했다면 올바른 CONSTRAINT_NAME과 CONSTRAINT를 찾을 수 있을 거예요.
# Here you have to copy the statements from SHOW CREATE TABLE output
ALTER TABLE <TABLE> ADD CONSTRAINT `<CONSTRAINT_NAME>` <CONSTRAINT>
이렇게 하면 데이터베이스가 새 Airflow 버전으로 마이그레이션할 수 있는 상태가 돼요.
업그레이드 후 경고 (Post-upgrade warnings)
보통은 airflow db migrate 명령을 성공적으로 실행하기만 하면 끝이에요. 하지만 어떤 경우 마이그레이션이 데이터베이스에서 오래되고 낡았으며 아마 잘못된 데이터를 발견해 별도의 테이블로 옮기기도 해요. 이 경우 webserver UI에서 발견된 데이터에 대한 경고가 표시될 수 있어요.
보게 될 전형적인 메시지:
Airflow found incompatible data in the
table in the metadatabase, and has moved them to during the database migration to upgrade. Please inspect the moved data to decide whether you need to keep them, and manually drop the table to dismiss this warning.
이런 메시지를 보면 일부 데이터가 손상된 것이므로, 그 데이터를 유지할지 삭제할지 결정하기 위해 검사해야 해요. 대부분 그 데이터는 일부 버그에서 남은 손상된 데이터이므로 안전하게 삭제할 수 있어요 — 그 데이터는 Airflow에서 어떻게든 보이거나 유용하지 않기 때문이에요. 하지만 감사 또는 역사적 이유로 특별히 필요하다면 어딘가에 저장해 둘 수도 있어요. 특별한 이유가 없다면 데이터를 삭제하는 것이 최선이에요.
데이터를 검사하고 삭제하는 방법은 다양해요. 자체 도구(종종 데이터베이스 객체를 보여주는 그래픽 도구)로 데이터베이스에 직접 접근할 수 있다면, 그런 도구로 테이블을 드롭하거나 이름을 바꾸거나 다른 데이터베이스로 옮길 수 있어요. 그런 도구가 없다면 airflow db shell 명령을 사용할 수 있어요 — 이 명령은 데이터베이스용 db shell 도구로 들어가게 해서 테이블을 검사하고 삭제할 수 있게 해줘요.
Kubernetes에서 테이블을 드롭하는 방법:
- Airflow pod 중 하나(webserver나 scheduler)에 exec로 들어가요:
kubectl exec -it <your-webserver-pod> python - python 셸에서 다음 명령을 실행해요:
from airflow.settings import Session
session = Session()
session.execute("DROP TABLE _airflow_moved__2_2__task_instance")
session.commit()
예시에서 <table>을 경고 메시지에 표시된 실제 테이블 이름으로 바꿔 주세요.
테이블 검사하기:
SELECT * FROM <table>;
테이블 삭제하기:
DROP TABLE <table>;
마이그레이션 모범 사례
데이터베이스 크기와 실제 마이그레이션에 따라 마이그레이션에 꽤 오랜 시간이 걸릴 수 있어요. 오랜 히스토리와 큰 데이터베이스가 있다면 먼저 데이터베이스 사본을 만들고 테스트 마이그레이션을 수행해 마이그레이션이 얼마나 걸릴지 평가하는 것을 권장해요. 일반적으로 "Major" 업그레이드는 새 기능 추가가 때로 데이터베이스 재구조화를 요구하므로 더 오래 걸릴 수 있어요.