[Airflow] Airflow Executor 정리

CeleryExecutor

• 의존성 설치
  • pip install 'apache-airflow[celery]'
  • 그외 RabbitMQ, Redis 등 Celery 백엔드에 따라 redis, librabbitmq 등 추가 의존성 설치

• Broker 를 이용한 task 관리
  • scheduler 는 broker 로 task 를 push
  • worker 는 broker 에서 task 를 pull 하여 실행
    • worker 는 계속 떠있는 상태에서 CeleryWorker 의 subprocess 로 task 를 실행하는 방식
  • broker 로는 RabbitMQ, Redis 등 선택

• worker node 가 여럿 있을 경우 task 분산 가능
• worker 노드에 필요한 조건들
  • 각 worker 에는 필요한 의존성 설치 필요
    • airflow 실행을 위해 airflow cli 설치 및 path 설정 필요
    • HiveOperator 를 사용할 경우 hive cli, MySqlOperator 를 사용할 경우 PYTHONPATH 가 잡혀있어야하는 등
  • DAGS_FOLDER 를 통해 DAG 파일 접근
    • worker 마다 git 을 주기적으로 pull 하는 방식 권장
  • cluster 내 모든 node 의 airflow config 가 동일해야함
    • 그렇지 않을 경우 오류 발생 가능
• 동시에 task 를 많이 실행할 경우 딜레이 발생할 수 있음

KubernetesExecutor

• 의존성 설치

  • pip install 'apache-airflow[cncf.kubernetes]'

• airflow scheduler 에서 프로세스로 실행

  • scheduler 자체는 k8s 에서 실행되지 않아도 되지만, k8s 클러스터 접근 권한은 필요

• metadata database 로 RDBMS 필요

  • sqllite 는 불가 (파일 기반이므로 분산된 다수 pod 에서 접근해야하는 상황에 부적합)

• KubernetesExecutor 를 위한 worker namespace, worker image 와 같은 정보들을 config 에 설정 필요

• executor 는 계속 떠있고, task 마다 worker pod 가 실행되는 방식

• config

  • airflow.cfg 의 kubernetes_executor 섹션 내에 pod template 을 위한 pod_template_file 설정 필요
    • base:
    • metadata.name : template name

---
apiVersion: v1
kind: Pod
metadata:
  name: placeholder-name
spec:
  containers:
    - env:
        - name: AIRFLOW__CORE__EXECUTOR
          value: LocalExecutor
        # Hard Coded Airflow Envs
        - name: AIRFLOW__CORE__FERNET_KEY
          valueFrom:
            secretKeyRef:
              name: RELEASE-NAME-fernet-key
              key: fernet-key
        - name: AIRFLOW__DATABASE__SQL_ALCHEMY_CONN
          valueFrom:
            secretKeyRef:
              name: RELEASE-NAME-airflow-metadata
              key: connection
        - name: AIRFLOW_CONN_AIRFLOW_DB
          valueFrom:
            secretKeyRef:
              name: RELEASE-NAME-airflow-metadata
              key: connection
      image: dummy_image
      imagePullPolicy: IfNotPresent
      name: base
      volumeMounts:
        - mountPath: "/opt/airflow/logs"
          name: airflow-logs
        - mountPath: /opt/airflow/airflow.cfg
          name: airflow-config
          readOnly: true
          subPath: airflow.cfg
  restartPolicy: Never
  securityContext:
    runAsUser: 50000
    fsGroup: 50000
  serviceAccountName: "RELEASE-NAME-worker-serviceaccount"
  volumes:
    - emptyDir: {}
      name: airflow-logs
    - configMap:
        name: RELEASE-NAME-airflow-config
      name: airflow-config

CeleryExecutor, KubernetesExecutor 비교

• CeleryExecutor 는 별도로 Redis 가 필요하지만, KubernetesExecutor 는 별도 component 불필요
  • k8s cluster 접근은 필요
• Celery Worker를 사용하면 task 가 대기열에 추가될 때 worker pod가 이미 실행 중 -> letency 감소
  • 여러 task 가 하나의 pod 에서 실행되므로, memory 등 리소스 사용량에 주의 필요
• 장기적으로 실행하는 task 가 있을 경우 KubernetesExecutor 가 유용
• CeleryKubernetesExecutor 를 사용할경우 celery 와 kubernetes executor 둘 모두 사용 가능
  • 기본 celery worker 로 실행시키지만, kubernetes queue 로 보내 실행하는 것도 가능

reference

• Airflow Docs - KubernetesExecutor
• Airflow Docs - Celery Executor