[Data Platform] Airflow 구축 가이드 (feat. Spark on Kubernetes)

---

Intro

 

안녕하세요. 환이s입니다 👋

 

지난 글에서는 데이터 레이크하우스 플랫폼의

네 번째 스택으로 Trino + Hive Metastore + PostgreSQL 3계층 구조로

Delta Lake SQL 조회 환경을 구축하는 과정을 정리해 봤습니다.

 

이번 글에서는 다섯 번째 스택,

Airflow를 Kubernetes 환경에 구축하는 과정을 정리해보려고 합니다.

 

Spark Streaming으로 처리된 데이터는

Delta Lake 포맷으로 Ceph S3에 쌓이게 됩니다.

 

이 흐름을 주기적으로 실행하고 스케줄링하려면

워크플로우 오케스트레이터가 필요한데,

 

그 역할을 담당하는 것이 바로 Airflow입니다.

 

 

이번 글에서는

• Helm으로 Airflow를 Kubernetes 위에 설치하는 방법
• CeleryExecutor + Redis 기반 분산 실행 구조
• FileBrowser로 DAG 파일을 Web UI에서 직접 관리하는 방법
• SparkKubernetesOperator로 Spark Job을 K8s에 제출하는 방법
• PythonOperator로 S3 이벤트 로그를 파티셔닝하고 Trino와 동기화하는 방법

까지 순서대로 확인해 보겠습니다.

---

1. Airflow 아키텍처 개요

 

본격적인 설치에 앞서,

이번 구성의 전체 아키텍처를 먼저 이해하는 것이 중요합니다.

 

FileBrowser (Web UI)
      │  DAG 파일 업로드/편집
      ▼
CephFS PVC (airflow-helm-dags)
      │  공유 볼륨 마운트
      ▼
Airflow Scheduler ──► Airflow Worker (CeleryExecutor)
                              │
                    SparkKubernetesOperator
                              │
                    SparkApplication YAML 제출
                              ▼
                    Spark Driver / Executor Pod
                              │
                    Ceph S3 (Delta Lake) 적재
                              │
                    Trino 파티션 동기화

 

각 컴포넌트의 역할은 아래와 같습니다.

 

컴포넌트	역할
Airflow Scheduler	DAG 스케줄 감지 및 Task 실행 지시
Airflow Worker	CeleryExecutor 기반으로 Task를 실제 실행
Redis	Celery 브로커 역할 — Scheduler와 Worker 간 Task 큐
CephFS PVC	Scheduler, Worker, Spark Pod가 공유하는 DAG 파일 저장소
FileBrowser	Web UI에서 DAG 파일을 직접 업로드/편집

 

즉, 여기서 확인할 수 있는 핵심은

DAG 파일이 저장된 CephFS PVC를 Scheduler, Worker, Spark Pod 모두가 동시에 마운트하기 때문에 ReadWriteMany 모드가 반드시 필요하다는 점입니다.

---

2. 설치 순서

 

이번 구축에서 사용하는 디렉토리 구조는 아래와 같습니다.

airflow/
├── helm/
│   ├── install.sh        # Helm 설치 스크립트
│   └── values.yaml       # Helm 커스텀 값
└── manifests/
    ├── 00_rbac/
    │   ├── 10_airflow-spark-role.yaml      # Spark 제어 권한 정의
    │   └── 20_airflow-spark-binding.yaml   # Worker SA에 권한 바인딩
    ├── 10_pvc/
    │   └── 10_airflow-pvc.yaml             # 로그 PVC (수동 생성)
    ├── 20_ingress/
    │   └── 10_airflow-ingress-webserver.yaml
    └── 30_filebrowser/
        ├── 10_airflow_file_browser.yaml    # FileBrowser Deployment
        └── 20_airflow-ingress-filebrowser.yaml

 

설치 순서는 반드시 아래 순서를 지켜야 합니다.

PVC 수동 생성 → RBAC 설정 → Helm 설치 → Ingress → FileBrowser

 

반드시 지켜야 하는 포인트가 1곳 있습니다.

로그 PVC(airflow-logs-manual-pvc)는 Helm 설치 전에 수동으로 먼저 생성해야 합니다.

 

values.yaml에서 existingClaim으로 참조하기 때문에,

Helm이 먼저 실행되면 PVC를 찾지 못해 오류가 발생합니다.

---

3. PVC 사전 생성 

 

Airflow 로그 PVC를 먼저 생성합니다.

# manifests/10_pvc/10_airflow-pvc.yaml
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: airflow-logs-manual-pvc
  namespace: airflow
spec:
  accessModes:
    - ReadWriteMany
  storageClassName: rook-cephfs
  resources:
    requests:
      storage: 5Gi

kubectl apply -f manifests/10_pvc/10_airflow-pvc.yaml

 

즉, 여기서 확인할 수 있는 핵심은

DAG 파일용 PVC는 Helm이 자동으로 생성하지만,

로그용 PVC는 반드시 Helm 설치 전에 수동으로 먼저 만들어야 한다는 점입니다.

---

4. RBAC 설정 (Spark 연동 권한)

 

Airflow Worker가 Kubernetes에 SparkApplication을 제출하려면

airflow-helm-worker ServiceAccount에 권한이 필요합니다.

# manifests/00_rbac/10_airflow-spark-role.yaml
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: airflow-spark-manager-role
  namespace: airflow
rules:
- apiGroups: ["sparkoperator.k8s.io"]
  resources: ["sparkapplications", "sparkapplications/status"]
  verbs: ["create", "get", "watch", "list", "update", "patch", "delete"]
- apiGroups: [""]
  resources: ["pods", "pods/log", "pods/status", "services",
              "configmaps", "persistentvolumeclaims"]
  verbs: ["create", "get", "watch", "list", "update", "patch",
          "delete", "deletecollection"]

# manifests/00_rbac/20_airflow-spark-binding.yaml
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: airflow-worker-spark-binding
  namespace: airflow
subjects:
- kind: ServiceAccount
  name: airflow-helm-worker
  namespace: airflow
roleRef:
  kind: Role
  name: airflow-spark-manager-role
  apiGroup: rbac.authorization.k8s.io

kubectl apply -f manifests/00_rbac/

 

---

5. Helm으로 Airflow 설치

 

Harbor 내부 레지스트리에서 OCI 방식으로 Helm Chart를 가져옵니다.

(Harbor가 없는 환경이라면 아래 공식 이미지를 확인해서 대체하시면 됩니다.)

bitnami/postgresql - Docker Image

 

filebrowser/filebrowser - Docker Image

 

redis - Official Image | Docker Hub

 

apache/airflow - Docker Image

 

# helm/install.sh
NAMESPACE=airflow
RELEASE_NAME=airflow
REGISTRY=harbor.local
CHART_VERSION=1.15.0

helm registry login $REGISTRY

helm upgrade --install $RELEASE_NAME \
  oci://$REGISTRY/helm-chart/airflow \
  --version $CHART_VERSION \
  --namespace $NAMESPACE \
  --create-namespace \
  -f values.yaml

values.yaml의 핵심 설정은 아래와 같습니다.

---

Executor 설정

executor: "CeleryExecutor"

CeleryExecutor는 Redis를 브로커로 사용해 Worker를 분산 실행합니다.

---

이미지 설정 (Harbor 내부 레지스트리)

(Harbor가 없는 환경이라면 위 공식 이미지를 확인해서 대체하시면 됩니다.)

images:
  airflow:
    repository: harbor.local/library/airflow
    tag: 2.10.5
  redis:
    repository: harbor.local/library/redis
    tag: '7.2'

---

스토리지 설정

dags:
  persistence:
    enabled: true
    storageClassName: "rook-cephfs"
    accessMode: ReadWriteMany   # Scheduler + Worker + Spark Pod 동시 마운트
    size: 2Gi

logs:
  persistence:
    enabled: true
    storageClassName: "rook-cephfs"
    existingClaim: airflow-logs-manual-pvc  # 3번에서 수동 생성한 PVC 참조

---

ArgoCD 연동 설정

migrateDatabaseJob:
  useHelmHooks: false
  jobAnnotations:
    "argocd.argoproj.io/hook": Sync

createUserJob:
  useHelmHooks: false
  jobAnnotations:
    "argocd.argoproj.io/hook": Sync

 

ArgoCD GitOps 배포 환경에서는

Helm Hook 대신 ArgoCD Hook으로 Job을 제어합니다.

 

useHelmHooks: false로 설정하지 않으면 ArgoCD와 충돌이 발생할 수 있습니다.

 

ArgoCD 자체 구축 및 설정에 대한 내용은

추후 ArgoCD 구축 가이드에서 더 자세하게 다루겠습니다.

ArgoCD를 사용하지 않는 환경이라면 이 설정은 생략하셔도 됩니다.

 

Airflow Pod가 Running 상태인지 확인합니다.

kubectl get pods -n airflow

 

정상적인 경우 아래처럼 보입니다.

NAME                                        READY   STATUS    RESTARTS   AGE
airflow-helm-scheduler-xxxxxxxxxx-xxxxx     1/1     Running   0          ...
airflow-helm-webserver-xxxxxxxxxx-xxxxx     1/1     Running   0          ...
airflow-helm-worker-xxxxxxxxxx-xxxxx        1/1     Running   0          ...
airflow-helm-redis-0                        1/1     Running   0          ...
airflow-helm-postgresql-0                   1/1     Running   0          ...

 

만약 Pod가 Running 상태가 되지 않는 경우 아래 명령어로 원인을 확인합니다.

# Pod 상태 상세 확인
kubectl describe pod <pod-name> -n airflow

# Pod 로그 확인
kubectl logs <pod-name> -n airflow

 

자주 발생하는 원인은 아래와 같습니다.

증상	원인	조치
Pending 상태 지속	PVC 바인딩 실패	kubectl get pvc -n airflow로 PVC 상태 확인
CrashLoopBackOff	로그 PVC를 찾지 못함	airflow-logs-manual-pvc 사전 생성 여부 확인
ImagePullBackOff	이미지 pull 실패	Harbor 로그인 상태 또는 공식 이미지 URL 확인

---

6. Ingress 설정

 

Airflow Webserver Ingress를 설정합니다.

# manifests/20_ingress/10_airflow-ingress-webserver.yaml
spec:
  ingressClassName: nginx
  rules:
  - host: airflow.<your-ip>.sslip.io
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: airflow-helm-webserver
            port:
              number: 8080

kubectl apply -f manifests/20_ingress/

---

7. FileBrowser 구성

 

DAG 파일을 Web UI에서 직접 올리고 편집할 수 있도록 FileBrowser를 구성합니다.

Airflow에서 DAG 파일을 반영하는 방법은 다양하지만 크게 두 가지가 있습니다.

방법	특징
Git Sync	GitOps 방식, PR/Commit 기반 반영
FileBrowser	Web UI에서 직접 업로드/편집, 즉시 반영

 

이번 구성에서는 FileBrowser를 선택했습니다.

DAG PVC(airflow-helm-dags)를 직접 마운트해서, 파일을 업로드하는 즉시 Scheduler가 감지합니다.

# manifests/30_filebrowser/10_airflow_file_browser.yaml
containers:
- name: explorer
  image: harbor.local/library/filebrowser:latest
  args: ["--database", "/config/database.db", "--noauth"]
  volumeMounts:
  - name: dags-data
    mountPath: /srv/airflow_dags   # DAG PVC 마운트
  - name: fb-config
    mountPath: /config
volumes:
- name: dags-data
  persistentVolumeClaim:
    claimName: airflow-helm-dags   # Airflow DAG PVC 공유
- name: fb-config
  persistentVolumeClaim:
    claimName: airflow-filebrowser-config-pvc

kubectl apply -f manifests/30_filebrowser/

즉, 여기서 확인할 수 있는 핵심은

FileBrowser가 Airflow의 DAG PVC를 직접 마운트하기 때문에 Web UI에서 파일을 올리는 즉시 Scheduler가 새 DAG를 인식한다는 점입니다.

---

8. DAG 작성 - Spark Job 제출

 

Airflow에서 Spark Job을 실행하는 방법은 두 가지로 나눌 수 있습니다.

---

방법 1. SparkKubernetesOperator (YAML 기반)

K8s에 SparkApplication YAML을 직접 제출하는 방식입니다.

# spark_s3_daily_job_v2.4
from airflow.providers.cncf.kubernetes.operators.spark_kubernetes import SparkKubernetesOperator

submit_spark = SparkKubernetesOperator(
    task_id='submit_spark_s3_job',
    namespace='airflow',
    application_file='yaml/spark-s3-reader.yaml',  # DAG PVC 내 YAML 참조
    kubernetes_conn_id='kubernetes_default',
    do_xcom_push=False,
    params={
        'interval_min': INTERVAL_MINUTES
    }
)

 

SparkApplication YAML에서 핵심 설정은 다음과 같습니다.

# yaml/spark-s3-reader.yaml (주요 부분)
spec:
  mainApplicationFile: "local:///opt/airflow/dags/scripts/filter_review.py"
  arguments:
    - "--start_date"
    - "{{ data_interval_start.in_timezone('Asia/Seoul').strftime('%Y-%m-%d %H:%M:%S') }}"
    - "--end_date"
    - "{{ data_interval_end.in_timezone('Asia/Seoul').strftime('%Y-%m-%d %H:%M:%S') }}"
  timeToLiveSeconds: 300  # 완료 후 5분 뒤 Pod 자동 삭제
  restartPolicy:
    type: Never
  sparkConf:
    # DAG PVC를 Driver/Executor Pod에도 마운트
    "spark.kubernetes.driver.volumes.persistentVolumeClaim.airflow-dags-vol.options.claimName": "airflow-helm-dags"
    "spark.kubernetes.driver.volumes.persistentVolumeClaim.airflow-dags-vol.mount.path": "/opt/airflow/dags"
    "spark.kubernetes.executor.volumes.persistentVolumeClaim.airflow-dags-vol.options.claimName": "airflow-helm-dags"
    "spark.kubernetes.executor.volumes.persistentVolumeClaim.airflow-dags-vol.mount.path": "/opt/airflow/dags"
    # 이벤트 로그 → Ceph S3 저장
    "spark.eventLog.enabled": "true"
    "spark.eventLog.dir": "s3a://spark-event-logs/"

 

즉, 여기서 확인할 수 있는 핵심은

Spark Driver/Executor Pod도 DAG PVC를 마운트해야

local:///opt/airflow/dags/scripts/filter_review.py 경로의 스크립트를 실행할 수 있다는 점입니다.

---

방법 2. PythonOperator (S3 직접 처리 + Trino 파티션 동기화)

Ceph S3의 Spark 이벤트 로그를 직접 읽어 처리하는 방식입니다.

# spark_s3_optimizer_v1.2
def add_regdt_and_partition(**context):
    s3 = boto3.client('s3',
        endpoint_url='http://<ceph-rgw-endpoint>',
        ...
    )

    # dt= 접두사가 이미 있는 파일은 스킵 (중복 처리 방지)
    if key.startswith('dt='):
        skipped_count += 1
        continue

    # 처리 완료 후 Trino 파티션 자동 동기화
    query = "CALL hive.system.sync_partition_metadata('spark_logs', 'raw_logs_partitioned', 'ADD')"

 

이 DAG는 30분 주기로 실행되며,

Spark Job이 쌓은 이벤트 로그를 날짜별 파티션 구조로 정리하고

Trino에서 바로 조회할 수 있도록 파티션을 동기화합니다.

 

즉, 여기서 확인할 수 있는 핵심은

Airflow가 단순 스케줄링에 그치지 않고,

S3 데이터 정리부터 Trino 파티션 동기화까지 파이프라인 전체를 자동화한다는 점입니다.

---

📝 마무리

이번 글에서는

Kubernetes 위에서 Airflow를 구축하고 Spark Job을 스케줄링하는 전체 파이프라인을 정리해 봤습니다.

 

정리하면,

• CeleryExecutor + Redis → Worker 분산 실행
• CephFS ReadWriteMany PVC → Scheduler, Worker, Spark Pod 간 DAG 파일 공유
• FileBrowser → Web UI에서 DAG 직접 업로드/편집, 즉시 반영
• SparkKubernetesOperator → YAML 기반으로 K8s에 Spark Job 제출
• PythonOperator + Trino 연동 → S3 이벤트 로그 파티셔닝 후 쿼리 자동 동기화

라고 볼 수 있습니다.

 

특히 이번 구축에서 주의해야 할 포인트는

• 로그 PVC는 Helm 설치 전에 수동 생성 필수
• DAG PVC는 반드시 ReadWriteMany — Spark Pod까지 동시 마운트 필요
• ArgoCD 환경에서는 useHelmHooks: false + ArgoCD Hook 설정 필수
• SparkApplication YAML에서 DAG PVC를 Driver/Executor 양쪽에 마운트해야 스크립트 실행 가능

이 4가지였습니다.

 

다음 편에서는 지금까지 구축한 전체 파이프라인을 모니터링하는 Prometheus + Grafana 구성을 다뤄보겠습니다. 🙌

 

궁금한 점이나 추가로 다뤄줬으면 하는 내용이 있다면 언제든지 편하게 말씀해 주세요! 😊