Airflow에서 KubernetesPodOperator로 kubectl image 쉽게 사용하는 방법
Airflow 파이프라인에서 kubectl을 이용해서 deployment rollout을 해야할 상황이 생겼습니다. 여러 방법을 고민해보다 KubernetesPodOperator와 Secret을 이용하면 쉽게 할 수 있어서 그 방법을 정리합니다.
Airflow
Airflow 병렬처리하기
병렬처리 고려 대상 Airflow에서 병렬차리할 때 고려해야할 대상이 크게 5가지 있습니다.
Airflow KubernetesPodOperator xcom 사용 시 발생하는 toomanyrequests 이슈
평소처럼 KubernetesPodOperator를 잘 활용하다 아래와 같은 오류 메세지가 발생했습니다.
Airflow S3에 있는 파일을 디스크를 안거치고 메모리로 가져오기
S3에 있는 파일을 디스크에 저장 후 작업을 진행하면 파일을 삭제하기 전에 예외 상황이 발생할 수 있고, 이렇게 되면 파일이 삭제되지 않고 디스크에 남아버립니다. 때문에 특별한 경우가 아니라면 디스크에 저장하지 않고 바로 가져와서 사용하는 것이 좋습니다.
Airflow MongoDB Provider의 replace_many
Airflow에서 MongoDB에 연결하기 위해 사용하는 MongoDB Provider에는 PyMongo에 없는 replace_many 함수가 있습니다. PyMongo에서 지원되는 관련 함수 목록은 아래와 같습니다.
Helm으로 설치한 Airflow에 MongoDB provider 추가
Helm으로 설치한 Airflow에 MongoDB connection을 추가하려했는데 MongoDB는 없었습니다. 공식 문서(링크)를 찾아보면 pip로 추가하라고 되어있지만 Helm으로 설치한 Airflow에 추가하려면 image를 새로 build해야합니다. 아래는 그 과정과 Mo...
Airflow Variables 활용하기 (+ Kubernetes 환경 변수로 추가)
Airflow Variables를 이용하면 Kubernetes의 ConfigMap, Secret와 유사하게 key, value를 관리할 수 있습니다.
Airflow 기간 내 Task clear (재실행)
Airflow Web UI에서도 task를 clear할 수 있지만, task의 양이 많거나 명령어를 이용해서 자동화하고 싶은 경우 아래 명령어를 이용해서 clear가 가능합니다.
KubernetesPodOperator에서 XCOM 사용하기
DockerOperator나 KubernetesPodOperator는 다른 Operator와 XCOM을 사용하는 방식이 조금 다릅니다. 여기서는 KubernetesPodOperator 기준으로 XCOM push와 pull 방법을 설명하겠습니다.
Airflow DummyOperator(EmptyOperator)
Airflow는 특정 Task에서 조건에 따라 skip하거나 멈추게하는 등의 동작을 하기 어렵습니다. 이를 해결하기 위해 branch로 분리해서 기존 flow가 진행되지 않도록 할 수 있는데, 이때 DummyOperator를 활용할 수 있습니다.
BranchPythonOperator로 Task 선택
BranchPythonOperator를 사용하면 여러 개의 Task 중 하나를 선택해서 flow를 진행할 수 있습니다. 조건에 따라 하나를 선택해서 사용하는 경우에 적용하면 유지 관리 측면에서도 효과를 볼 수 있습니다.
Airflow 분 단위 주기 설정
Airflow는 스트리밍 솔루션이 아니기 때문에 한 시간 이하의 주기로 동작시키는 경우를 많이 볼 수 없습니다. 그러나 주기만 줄여서 사용하고 싶다면 분 단위까지 낮춰서 사용할 수 있습니다.
KubernetesPodOperator를 이용해서 GPU 할당 받기
Airflow에서 딥러닝 서비스나 학습 파이프라인을 구축할 경우 필요한 만큼의 GPU만 할당해서 사용할 수 있도록 해야합니다. 그냥 사용하면 서버에 있는 전체 GPU를 잡아서 사용할 수 있습니다. 멀티 GPU 서버에서 단일 GPU만 할당받아서 사용하는 방법은 아래와 같습니다.
Airflow에서 PyTorch 사용 (KubernetesPodOperator)
Airflow를 Kubernetes에 올려놓고 사용중인데, PyTorch를 사용하기 위한 패키지들을 Airflow에 설치하고 싶지 않아 알아보던 중 KubernetesPodOperator를 이용해서 해결했습니다. Kubernetes에 설치하지 않았다면 DockerOperator를 ...
Airflow v2.2 이상 schedule 관련 용어
기존에 Airflow에서 schedule 처리를 진행할 때 execution_date, next_execution_date과 같은 변수를 사용해왔는데, 2.2버전부터 좀 더 명확한 용어로 변경되었습니다. 현재 최신 버전인 v2.3을 이용해도 호환성을 위해 아직 사용이 가능하지만, ...
Airflow 2.3 설치 (w. WSL2)
Airflow가 아직 윈도우를 지원하지 않기 때문에 WSL을 이용해서 설치하는 방법을 설치하는 겸 Ubuntu에서 설치하는 방법과 큰 차이는 없기 때문에 두 가지 방법을 비교하면서 설명하겠습니다.