데이터 파이프라인 프로젝트 #2 인프라 구성
아키텍처가 나온 후 인프라 구성에 들어갔다. 이전 글에 적어 둔 것처럼 k3s위에 airflow를 올려서 사용 할 예정이다.
k3s의 설치는 아주 간단하다. 내부적으로 설정을 잡아 줄 일이 있다면 복잡해지지만, 단순히 설치만 하고 사용 할 것이라면 설치 스크립트 하나로 설치가 가능하다.
Executor vs PodOperator
kubernetes에 airflow를 구성할 때 2가지 방식을 고려해 볼 수 있다. 제목에 있는 것 처럼 KubernetesExecutor와 KubernetesPodOperator를 사용하는 것이다. 두 방식의 차이는 결국 executor와 operator의 차이다.
Executor는 task의 실행 방법을 결정한다. 대표적인 설정은 Celery,Local,Kubernetes등이 있다. 각각의 executor는 설정 된 대로 Celery(python환경) 에서 task를 실행 할 것인지 Local에서 혹은 Kubernetes에서 실행할지를 결정하게 되는 것이다.
Operator는 task가 무엇을 할지 결정하게 된다. Bash,Python 등의 Operator가 있다. dag 정의시 operator를 결정하게 되는데, task가 어떤 작업을 하게 될 지 정하게 된다. 예를들어 BashOperator의 경우 bash 작업을 airflow의 dag에 추가하여 진행하게 되는 것이다.
Kubernetes Executor
나는 프로젝트에 KubernetesExecutor를 사용 할 것이다. airflow가 DAG을 실행하면 각각의 task가 pod로 실행되고 task 종료시 pod 또한 삭제 되는 방식으로 진행된다. 이 방식을 택한 이유는 다음과 같다.
KubernetesExecutor를 사용한 Airflow의 아키텍처를 살펴보자. 기존의 경우 Worker는 하나의 컴포넌트로써 종료되지 않고 pod로 실행되어 task를 처리하게 된다. 이때 작업을 하지 않는 순간까지도 Worker가 실행되는 자원을 낭비 할 가능성이 생기게 된다. 따라서 task의 처리 완료와 동시에 Worker pod를 종료시키는 KubernetesExecutor를 선택하게 되었다.
물론 task의 runtime에 pod가 생성되기 때문에 지연이 발생할 수 있다. 하지만 현재 프로젝트 규모가 크지 않기 때문에 지연 발생에 대해서는 크게 신경쓰지 않아도 되는 단계라고 판단하여 추후에 다시 고려하는 것으로 결정했다.
구축
helm + kustomize 를 사용하여 구축했다. DAG 관리의 경우 git sync를 사용하여 관리 편의성을 확보했다.
executor: "KubernetesExecutor"
dags:
gitSync:
enabled: true
repo: {{github address}}
branch: main
rev: HEAD
depth: 1
maxFailures: 0
subPath: "dags"
period: 5s
credentialsSecret: git-credentials
sshKeySecret: airflow-git-ssh-secret
containerName: git-sync
uid: 65533
실행후 1차 좌절을 맛봤다. 파이프라인의 실행은 잘 이뤄졌지만 log를 읽어오지 못했다.
data-pipeline-health-check-39cdam7n
*** Could not read served logs: HTTPConnectionPool(host='data-pipeline-health-check-39cdam7n', port=8793): Max retries exceeded with url: /log/dag_id=data_pipeline/run_id=scheduled__2025-02-17T01:01:00+00:00/task_id=health_check/attempt=1.log (Caused by NameResolutionError("<urllib3.connection.HTTPConnection object at 0x7bf772a6e540>: Failed to resolve 'data-pipeline-health-check-39cdam7n' ([Errno -2] Name or service not known)"))
마지막 task가 실패했지만, 그래도 앞의 2개 task는 성공했으니 log가 나와야 맞는 것 이라고 생각했지만 나오지 않았다.
여기서 고려해 본 사항은 2개가 있다. 1.pod 네트워크 문제 2.log 저장 볼륨 문제
1.pod 네트워크 문제
에러 로그에 HTTPConnectionPool과 NameResolutionError 를 보고 DNS의 문제일 수 있을 거라는 생각을 해 보았다.
2.log 저장 볼륨 문제
KubernetesExecutor를 사용하게 되면 worker pod가 작업 완료 후 제거되기 때문에 log를 따로 저장하지 않는다면 유실되어 읽을 수 없을 것 이라고 생각했다.
정답은 1,2번 모두에 있었다. 역시 pod가 사라지기 때문에 네트워크 Connection이 이뤄지지 않았고 log를 따로 볼륨에 저장하지 않았기 때문에 읽어올 것이 없었다. 그래서 pvc를 생성하고 연결해 보았다. 하지만 이번엔 airflow 자체가 실행되지 않았다. (이 문제는 추후에 따로 포스팅 하는 것으로 하겠다.) 시간을 너무 잡아먹고 있는 것 같아 우선 EmptyDir을 설정하여 log를 보관하는 것으로 했다.
여기까지 KubernetesExecutor를 사용하여 Airflow를 실행하는 단계까지 완료가 되었다.\