Airflow on Kubernetes

Operating Airflow in k8s executor

Airflow 아키텍처

Airflow on k8s

KubernetesExecutor에 대해

K8s 내에서 airflow 띄우기

Git Sync

KubernetesExecutor로 설정

Dynamic webserver 키가 아닌 static key 부여

Operating Airflow in k8s executor

Airflow 아키텍처

Airflow on k8s

KubernetesExecutor에 대해

K8s 내에서 airflow 띄우기

Git Sync

KubernetesExecutor로 설정

Dynamic webserver 키가 아닌 static key 부여

Airflow 아키텍처

Airflow on k8s

: 보통 쿠버네티스 위에 Airflow를 구동시키면 Airflow 컴포넌트들이 파드의 형태로 구동하게 된다.

•

구성이 간단함

•

서비스들이 단순히 파드로 변환된 것이므로 템플릿화하기 쉬움

BUT, 결국 구성요소들이 파드로 변환되기 때문에 계속해서 자원을 점유하게 된다.

: 다음과 같이 worker 노드가 계속해서 자원을 점유하고 있는 것을 확인할 수 있다. (심지어 DAG가 트리거가 되지 않았는데도!)

추가로, 이미지들로 쿠버네티스에서 airflow를 조성하다보니, 의존성 관리 및 확장성에 매우 큰 어려움이 생긴다.

KubernetesExecutor에 대해

: KubernetesExecutor은 계속해서 워커가 리소스를 점유하는 것이 아닌, 스케줄러에 의해서 동적으로 워커 노드가 생긴다.

•

Scheduler가 해당 시점에서 실행되어야 할 태스크를 찾는다.

•

Executor는 이 태스크를 실행하기 위해 워커 노드를 Pod 형태로 실행한다.

def as_pod(self):

    if self.kube_config.pod_template_file:
        return PodGenerator(pod_template_file=self.kube_config.pod_template_file).gen_pod()
    pod = PodGenerator(
        image=self.kube_config.kube_image,
        image_pull_policy=self.kube_config.kube_image_pull_policy or 'IfNotPresent',
        image_pull_secrets=self.kube_config.image_pull_secrets,
        volumes=self._get_volumes(),
        volume_mounts=self._get_volume_mounts(),
        init_containers=self._get_init_containers(),
        labels=self.kube_config.kube_labels,
        annotations=self.kube_config.kube_annotations,
        affinity=self.kube_config.kube_affinity,
        tolerations=self.kube_config.kube_tolerations,
        envs=self._get_environment(),
        node_selectors=self.kube_config.kube_node_selectors,
        service_account_name=self.kube_config.worker_service_account_name or 'default',
        restart_policy='Never'
    ).gen_pod()

    pod.spec.containers[0].env_from = pod.spec.containers[0].env_from or []
    pod.spec.containers[0].env_from.extend(self._get_env_from())
    pod.spec.security_context = self._get_security_context()

    return append_to_pod(pod, self._get_secrets())
Python
복사

•

만일 개별 설정된 pod 템플릿 파일이 있으면 이를 따르고, 아니라면 기본 kube_config에 따라 워커노드를 파드 형태로 생성한다.

•

그렇게 파드가 생성되면 파드 리스트에 해당 파드가 생성되며, 이 파드는 워커 노드의 역할을 수행하게 된다.

동적으로 워커노드를 관리하기 때문에 이에 따른 장점이 많다.

•

각종 의존성 패키지들을 구성할 필요 없이 K8sExecutor, k8sPodOperator만 사용하면 된다.

•

컨테이너 이미지 기반 운영으로 인한 의존성 확인 작업이 필요 없어지므로 유지 보수 비용이 절감된다.

•

태스크가 실행될 때만 동적으로 워커 노드를 생성하므로 효율적인 자원 관리가 이루어진다.

•

K8sPodOperator을 통해 DAG코드를 템플릿화 할 수 있다.

⇒ 그러나 굉장히, 굉장히…구성이 복잡하다.

하지만? 우린 이걸 오늘 구성해볼것이다.

K8s 내에서 airflow 띄우기

: k8s 내에서 airflow를 설치하는 방법이다. 우선 minikube가 필요하니, minikube부터 설치 (설치법 생략)

minikube start
😄  Darwin 13.4 (arm64) 의 minikube v1.30.1
🎉  minikube 1.31.2 이 사용가능합니다! 다음 경로에서 다운받으세요: https://github.com/kubernetes/minikube/releases/tag/v1.31.2
💡  해당 알림을 비활성화하려면 다음 명령어를 실행하세요. 'minikube config set WantUpdateNotification false'
✨  자동적으로 docker 드라이버가 선택되었습니다
📌  Using Docker Desktop driver with root privileges
👍  minikube 클러스터의 minikube 컨트롤 플레인 노드를 시작하는 중
🚜  베이스 이미지를 다운받는 중 ...
🔥  Creating docker container (CPUs=2, Memory=7903MB) ...
🐳  쿠버네티스 v1.26.3 을 Docker 23.0.2 런타임으로 설치하는 중
    ▪ 인증서 및 키를 생성하는 중 ...
    ▪ 컨트롤 플레인이 부팅...
    ▪ RBAC 규칙을 구성하는 중 ...
🔗  Configuring bridge CNI (Container Networking Interface) ...
    ▪ Using image gcr.io/k8s-minikube/storage-provisioner:v5
🔎  Kubernetes 구성 요소를 확인...
🌟  애드온 활성화 : storage-provisioner, default-storageclass
🏄  끝났습니다! kubectl이 "minikube" 클러스터와 "default" 네임스페이스를 기본적으로 사용하도록 구성되었습니다.
Bash
복사

그후 helm을 설치해야 한다. 있으면 이 부분은 생략

(root user)

curl -fsSL -o get_helm.sh https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3
chmod 700 get_helm.sh
./get_helm.sh


which helm
/usr/local/bin/helm

helm의 위치 찾기: 먼저, root 사용자로 로그인하여 helm의 정확한 위치를 찾습니다.

sudo su
which helm
이 명령은 helm의 경로를 출력합니다. 예를 들어, 출력 결과가 /usr/local/bin/helm이라면, 이 경로를 기록해 두세요.

PATH 환경 변수 수정: 이제 사용자로 돌아가서 PATH 환경 변수에 helm의 위치를 추가합니다. 
사용자의 셸 설정 파일 (예: .bashrc 또는 .zshrc)을 편집해야 합니다.

exit  # root 사용자에서 나오기

(user)
nano ~/.bashrc  # 또는 사용하는 셸에 맞는 설정 파일을 엽니다

파일의 마지막에 다음 라인을 추가합니다 (여기서는 helm이 /usr/local/bin/helm에 위치한다고 가정합니다):]
export PATH=$PATH:/usr/local/bin
변경사항을 저장하고 편집기를 종료합니다.

변경사항 적용: 변경사항을 적용하기 위해 셸 설정 파일을 다시 로드합니다.

source ~/.bashrc  # 또는 사용하는 셸에 맞는 설정 파일
helm 실행 확인: 이제 helm 명령어를 실행하여 변경사항이 올바르게 적용되었는지 확인합니다.

helm version
Bash
복사

Helm Chart for Apache Airflow — helm-chart Documentation

Guides

https://airflow.apache.org/docs/helm-chart/stable/index.html

helm 레포지토리를 추가한 다음, airflow 설치

helm repo add apache-airflow https://airflow.apache.org
SQL
복사

airflow 1.11.0 · apache-airflow/apache-airflow

The official Helm chart to deploy Apache Airflow, a platform to programmatically author, schedule, and monitor workflows

https://artifacthub.io/packages/helm/apache-airflow/airflow

# values.yaml
airflowVersion: "2.6.2"
SQL
복사

helm upgrade --install airflow apache-airflow/airflow --namespace airflow --create-namespace -f my-values.yaml
Bash
복사

포트 포워딩:

airflow 네임스페이스의 airflow-webserver 서비스의 8080 포트를 로컬 머신의 20000 포트로 포워딩합니다.

kubectl port-forward svc/airflow-webserver 8080:8080 -n airflow

kubectl port-forward svc/airflow-webserver 20000:8080 -n airflow
Bash
복사

localhost:8080 or 20000으로 접속하면 다음과 같이 airflow UI가 뜬다!

ID: admin, PW: admin

k8s 위에 airflow를 띄우면 다음과 같은 구조를 형성한다.

kubectl get pods -n airflow

NAME                                 READY   STATUS    RESTARTS   AGE
airflow-postgresql-0                 1/1     Running   0          2m12s
airflow-redis-0                      1/1     Running   0          2m12s
airflow-scheduler-68d6b6cdd4-z2ngc   2/2     Running   0          2m12s
airflow-statsd-77685bcd45-t6pfj      1/1     Running   0          2m12s
airflow-triggerer-0                  2/2     Running   0          2m12s
airflow-webserver-bf8bbc9cb-qgd85    1/1     Running   0          2m12s
airflow-worker-0                     2/2     Running   0          2m12s
Bash
복사

하지만, airflow를 container로 띄운 만큼, 언제 재시작해도 전혀 이상하지가 않다. 따라서, git sync를 통해 레포지토리에서 DAG 코드를 지속적으로 가져올 수 있다.

⇒ 여기서 DAG Directory가 자신의 Git Repository가 되는 것이다.

Git Sync

아래 과정은 private repo를 기준으로 진행된다.

참고로 public repo의 경우에는, 깃 주소만 입력하면 된다. 별도의 Deploy key는 필요하지 않다.

11708

issues

: DAG를 저장할 Private Repository를 만든다.

ssh-keygen -t rsa -b 4096 -C "your_email@example.com"
Bash
복사

: .ssh 디렉토리 내에 airflow_ssh_key 라는 이름으로 ssh key를 생성한다. 이 때, passphrase를 입력하지 않고 빈칸으로 둔다.

pbcopy < airflow_ssh_key.pub
Bash
복사

: 그후 클립보드에 ssh_key 퍼블릭키를 복사한다.

:복사한 퍼블릭키를 레포지토리의 키에 복붙한 후, 추가한다.

kubectl create secret generic \ 
	airflow-git-ssh-secret \
  --from-file=gitSshKey=/Users/username/.ssh/airflow_ssh_key \
  --from-file=id_ed25519.pub=/Users/username/.ssh/airflow_ssh_key.pub \
  -n airflow
Bash
복사

그 후, 만든 키와 퍼블릭키를 k8s 클러스터 안에 secret으로 추가한다.

cd ~/.ssh
vi values.yaml
Bash
복사

이제, airflow 설정 yaml파일에 secret을 등록하기 위해 values.yaml파일을 수정한다.

/# Git sync
airflowVersion: "2.6.2"

extraEnv: |
    - name: "AIRFLOW__CORE__PLUGINS_FOLDER"
      value: "/opt/airflow/dags/repo/plugins"

# Git sync
dags:
  persistence:
    # Enable persistent volume for storing dags
    enabled: true #변경 1
    # Volume size for dags
    size: 10Gi
    # If using a custom storageClass, pass name here
    storageClassName:
    # access mode of the persistent volume
    accessMode: ReadWriteOnce
    ## the name of an existing PVC to use
    existingClaim:
  gitSync:
    enabled: true #변경 2
    repo: git@github.com:xxx/xxxx.git #변경 3
    branch: main #변경 4
    rev: HEAD
    depth: 1
    # the number of consecutive failures allowed before aborting
    maxFailures: 0
    # subpath within the repo where dags are located
    # should be "" if dags are at repo root
    subPath: ""

    credentialsSecret: git-credentials

    sshKeySecret: airflow-git-ssh-secret #변경 5

    wait: 10
    containerName: git-sync
    uid: 65533

    # When not set, the values defined in the global securityContext will be used
    securityContext: {}
    #  runAsUser: 65533
    #  runAsGroup: 0

    extraVolumeMounts: []
    env: []
    resources: {}
    #  limits:
    #   cpu: 100m
    #   memory: 128Mi
    #  requests:
    #   cpu: 100m
    #   memory: 128Mi
YAML
복사

: 위에서 #변경에 해당하는 부분을 다음과 같이 수정한 후, 저장한다.

helm upgrade --install airflow apache-airflow/airflow -n airflow -f my-values.yaml
Bash
복사

: value.yaml을 수정하여 helm upgrade를 시도하면…?

Error: UPGRADE FAILED: cannot patch "airflow-scheduler" with kind Deployment: Deployment.apps "airflow-scheduler" is invalid: [spec.template.spec.containers[1].volumeMounts[1].name: Not found: "git-sync-ssh-key", spec.template.spec.initContainers[1].volumeMounts[1].name: Not found: "git-sync-ssh-key"]
Bash
복사

실패가 떠버린다. 오류를 수정해보자.

vi values.yaml
/# Airflow scheduler settings
Bash
복사

다시 values.yaml로 들어간다.

scheduler:
  extraVolumes:
    - name: git-sync-ssh-key
      secret:
        secretName: airflow-git-ssh-secret
YAML
복사

/# Airflow scheduler settings 를 입력해 키워드를 찾은 후, 위의 scheduler 코드를 추가한다.

helm upgrade --install airflow apache-airflow/airflow -n airflow -f values.yaml
Bash
복사

다시 upgrade를 하면, 성공적으로 될 것이다.

from datetime import datetime, timedelta
from textwrap import dedent

from airflow import DAG

from airflow.operators.bash import BashOperator
from airflow.operators.python_operator import PythonOperator

from datetime import datetime 
def test():
    print(datetime.now())

    
with DAG(
    'tutorial',
    # These args will get passed on to each operator
    # You can override them on a per-task basis during operator initialization
    default_args={
        'depends_on_past': False,
        'email': ['airflow@example.com'],
        'email_on_failure': False,
        'email_on_retry': False,
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
    },
    description='A simple tutorial DAG',
    schedule_interval='1 * * * *',
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=['example'],
) as dag:

    # t1, t2 and t3 are examples of tasks created by instantiating operators
    t1 = PythonOperator(
        task_id='print_date',
        python_callable=test,
    )

    t2 = BashOperator(
        task_id='sleep',
        depends_on_past=False,
        bash_command='sleep 5',
        retries=3,
    )
    t1.doc_md = dedent(
        """\
    #### Task Documentation
    You can document your task using the attributes `doc_md` (markdown),
    `doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets
    rendered in the UI's Task Instance Details page.
    ![img](http://montcs.bloomu.edu/~bobmon/Semesters/2012-01/491/import%20soul.png)
    """
    )

    dag.doc_md = __doc__  # providing that you have a docstring at the beginning of the DAG
    dag.doc_md = """
    This is a documentation placed anywhere
    """  # otherwise, type it like this
    templated_command = dedent(
        """
    {% for i in range(5) %}
        echo "{{ ds }}"
        echo "{{ macros.ds_add(ds, 7)}}"
    {% endfor %}
    """
    )

    t3 = BashOperator(
        task_id='templated',
        depends_on_past=False,
        bash_command=templated_command,
    )

    t1 >> [t2, t3]
Python
복사

: git-sync가 제대로 되었는지 확인하기 위해 dag파일을 만들어 git repository에 푸쉬한 후, 확인해보자.

짜잔. 다음과 같이 DAG 리스트가 업데이트 된 것을 볼 수 있다.

BUT…반쪽짜리 k8s airflow…?

원래 kubernetes를 통해 airflow를 구동하면, DAG가 트리거 될 경우에만 워커 노드가 생성되어 리소스를 할당받고, 태스크가 완료되면 리소스를 반환해야 되는데, 워커가 계속 돌아가고 있다. KubernetesExecutor로 설정되어있지 않아서 그렇다!

KubernetesExecutor로 설정

vi values.yaml
Bash
복사

: values.yaml에 다시 들어가서 executor를 변경한다.

이렇게 되면 평상시에는 워커노드가 생성되어있지 않다.

다음과 같이 DAG를 트리거 하면, DAG 아이디에 해당하는 워커 노드가 생성되어 태스크를 수행한다.

그 후, 태스크가 완료되면 파드는 종료되고 리소스를 반환한다.

최종적인 overriding values.yml은 아래와 같다

private repo 기준

airflowVersion: "2.6.2"

extraEnv: |
    - name: "AIRFLOW__CORE__PLUGINS_FOLDER"
      value: "/opt/airflow/dags/repo/airflow/plugins"

executor: "KubernetesExecutor"


dags:
  persistence:
    # Enable persistent volume for storing dags
    enabled: true #변경 1
    # Volume size for dags
    size: 10Gi
    # If using a custom storageClass, pass name here
    storageClassName:
    # access mode of the persistent volume
    accessMode: ReadWriteOnce
    ## the name of an existing PVC to use
    existingClaim:

  gitSync:
    enabled: true
    repo: git@github.com:xxx/xxxx.git
    branch: main
    rev: HEAD
    depth: 1
    # the number of consecutive failures allowed before aborting
    maxFailures: 0
    # subpath within the repo where dags are located
    # should be "" if dags are at repo root
    subPath: "airflow/dags"

    credentialsSecret: git-credentials

    sshKeySecret: airflow-git-ssh-secret #변경 5

    wait: 10
    containerName: git-sync
    uid: 65533

    # When not set, the values defined in the global securityContext will be used
    securityContext: {}
    #  runAsUser: 65533
    #  runAsGroup: 0

    extraVolumeMounts: []
    env: []
    resources: {}
    #  limits:
    #   cpu: 100m
    #   memory: 128Mi
    #  requests:
    #   cpu: 100m
    #   memory: 128Mi

scheduler:
  extraVolumes:
    - name: git-sync-ssh-key
      secret:
        secretName: airflow-git-ssh-secret
SQL
복사

Dynamic webserver 키가 아닌 static key 부여

Kubernetes환경에서 Airflow를?!?!

Kubernetes환경에서 Airflow를?!?! 요즘 가장 핫한 Workflow Tool은 Airflow입니다. python으로 개발되어서 설치도 PyPl로 간편하게 설치할 수 있습니다. 하지만 디테일하게 사용하려면 Metastore, Celery Worker 등 설정할게 많습니다. 이러던 중 Kubernetes에 Airflow를 설치해야 될 업무를 맡게 되었습니다. 이번 포스팅에서는 Airflow를 Kubernetes 환경에 설치하는 것을 정리해 보았습니다. Helm으로 Airflow 설치 Airflow에서 공식 Helm을 제공해 주기 때문에 해당 Helm으로 설치를 진행하였습니다. 다른 버전인 User-community Helm도 존재하니 참고해 주시길 바랍니다. Airflow Helm Chart를..

https://developnote-blog.tistory.com/176

python3 -c 'import secrets; print(secrets.token_hex(16))'
f83bab1deefae2199d27d2621762c30b
SQL
복사

Airflow에서는 webserver로 flask를 사용한다

flask는 session id를 관리하기 위해 필요한데, 난수를 만들어서 k8s의 secret object로 생성을 해야한다

kubectl create secret generic \
airflow-webserver-secret \
--from-literal="webserver-secret-key=f83bab1deefae2199d27d2621762c30b" \
-n airflow

secret/airflow-webserver-secret created
SQL
복사

airflowVersion: "2.6.2"

extraEnv: |
    - name: "AIRFLOW__CORE__PLUGINS_FOLDER"
      value: "/opt/airflow/dags/repo/airflow/plugins"

# 추가
webserverSecretKey: webserver-secret-key
webserverSecretKeySecretName: airflow-webserver-secret

executor: "KubernetesExecutor"


dags:
  persistence:
    # Enable persistent volume for storing dags
    enabled: true #변경 1
    # Volume size for dags
    size: 10Gi
    # If using a custom storageClass, pass name here
    storageClassName:
    # access mode of the persistent volume
    accessMode: ReadWriteOnce
    ## the name of an existing PVC to use
    existingClaim:

  gitSync:
    enabled: true
    repo: https://github.com/xxx/xxx.git
    branch: main
    rev: HEAD
    depth: 1
    # the number of consecutive failures allowed before aborting
    maxFailures: 0
    # subpath within the repo where dags are located
    # should be "" if dags are at repo root
    subPath: "airflow/dags"

    # credentialsSecret: git-credentials

    # sshKeySecret: airflow-git-ssh-secret #변경 5

    wait: 10
    containerName: git-sync
    uid: 65533

    # When not set, the values defined in the global securityContext will be used
    securityContext: {}
    #  runAsUser: 65533
    #  runAsGroup: 0

    extraVolumeMounts: []
    env: []
    resources: {}
    #  limits:
    #   cpu: 100m
    #   memory: 128Mi
    #  requests:
    #   cpu: 100m
    #   memory: 128Mi

# 추가
webserver:
  service:
    type: NodePort
    ## service annotations
    ports:
      - name: airflow-ui
        port: "{{ .Values.ports.airflowUI }}"
        targetPort: "{{ .Values.ports.airflowUI }}"
        nodePort: 31151
SQL
복사

helm upgrade --install airflow apache-airflow/airflow -n airflow -f my-values.yaml
SQL
복사

다시 배포를 진행해본다.

$ minikube ip

192.168.49.2

$ kubectl get svc -n airflow
NAME                    TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)             AGE
airflow-postgresql      ClusterIP   10.100.22.149   <none>        5432/TCP            30m
airflow-postgresql-hl   ClusterIP   None            <none>        5432/TCP            30m
airflow-statsd          ClusterIP   10.107.91.48    <none>        9125/UDP,9102/TCP   30m
airflow-triggerer       ClusterIP   None            <none>        8794/TCP            30m
airflow-webserver       NodePort    10.100.214.92   <none>        8080:31151/TCP      30m

# 아래의 주소로 접근
http://192.168.49.2:31151
SQL
복사

log 이슈

dag가 잘 실행은 되는데 아래와 같은 이슈가 발견되었다.

Could not read served logs: [Errno -2] Name or service not known

웹서버 파드의 로그를 한번 살펴보자.

$ kubectl logs airflow-webserver-78866bff9b-cbrx4 -n airflow

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/file_task_handler.py", line 505, in _read_from_logs_server
    response = _fetch_logs_from_service(url, rel_path)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/file_task_handler.py", line 93, in _fetch_logs_from_service
    headers={"Authorization": signer.generate_signed_token({"filename": log_relative_path})},
  File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_api.py", line 201, in get
    trust_env=trust_env,
  File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_api.py", line 110, in request
    follow_redirects=follow_redirects,
  File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_client.py", line 821, in request
    return self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_client.py", line 912, in send
    history=[],
  File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_client.py", line 939, in _send_handling_auth
    history=history,
  File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_client.py", line 973, in _send_handling_redirects
    response = self._send_single_request(request)
  File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_client.py", line 1009, in _send_single_request
    response = transport.handle_request(request)
  File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_transports/default.py", line 218, in handle_request
    resp = self._pool.handle_request(req)
  File "/usr/local/lib/python3.7/contextlib.py", line 130, in __exit__
    self.gen.throw(type, value, traceback)
  File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_transports/default.py", line 77, in map_httpcore_exceptions
    raise mapped_exc(message) from exc
httpx.ConnectError: [Errno -2] Name or service not known
SQL
복사

Manage logs — helm-chart Documentation

Guides

https://airflow.apache.org/docs/helm-chart/stable/manage-logs.html

Airflow: Unable to get persistent logs using KubernetesExecutor and PV (official helm chart)

I feel a bit like an idiot but I cannot seem to get the logging working on persistent volumes when using the KubernetesExecutor and the freshly released official Helm chart. After creating a simple...

https://stackoverflow.com/questions/68225077/airflow-unable-to-get-persistent-logs-using-kubernetesexecutor-and-pv-official

Airflow KubernetesExecutor, logs do not show up in UI until after executor pods complete

I have started using KubernetesExecutor and I have set up a PV/PVC with an AWS EFS to store logs for my dags. I am also using s3 remote logging. All the logging is working perfectly fine after a dag

https://stackoverflow.com/questions/65929595/airflow-kubernetesexecutor-logs-do-not-show-up-in-ui-until-after-executor-pods

https://www.ibm.com/docs/en/dobd?topic=integration-airflow-live-logs-in-kubernetes

이유는 Airflow의 로그 지속성과 연관 있다.

kubernetes executor를 사용하게 되면 동적으로 worker를 사용하게 되는데,

이 때 로깅을 워커에 하게 되고, 워커가 사라지게 되면 해당 service 에 대해서 로그를 가져오지 못하게 되는 문제가 생긴다

이 문제를 해결하기 위해서는 로깅 스토리지를 s3와 같이 분리하거나, 추가적인 외부 볼륨(PVC)을 잡아야한다

Airflow에서 로그 지속성(persistence)을 활성화하면, Airflow 컴포넌트들이 로그를 동일한 볼륨에 기록하게 된다. 이를 위해 ReadWriteMany 접근 모드를 가진 PersistentVolumeClaim이 프로비저닝된다.

아래에 추가적인 해결 방법을 명시해보도록 하겠다.

values.yaml 아래 명시를 하자.

logs:
  persistence:
    enabled: true
SQL
복사

즉, 기본 로그가 false인데, true 바꿔줘야 Persistence volume에 Pod log를 남긴다.

위와 같이 설정 후 helm을 재설치하자.

위와 같이 정상적으로 로그를 볼 수 있다.

requirements.txt 설치

Airflow On K8s (Kubernetes) - Provider Dependency 설치하기

Provider dependency를 설치하기 위해서는 Custom docker image를 Build 해야 한다.데이터 분석 및 처리를 위해 많이 사용하는 Spark dependency를 설치해보자.requirements.txtDockerfiledocker build

https://velog.io/@jskim/Airflow-On-K8s-Kubernetes-Provider-Dependency-설치하기

참고 자료

Helm Chart for Apache Airflow — helm-chart Documentation

Guides

https://airflow.apache.org/docs/helm-chart/stable/index.html

EKS 위에 Airflow v2.5.3 Helm 배포하기

개요 EKS 클러스터 위에서 helm chart apache-airflow/airflow v1.9.0으로 airflow v2.5.3를 간단히 배포해보자. (굳이 EKS가 아니어도 큰 틀은 같음.) 구체적인 구현 내용은 아래와 같다. Service는 aws-load-balancer-controller로 LoadBalancer 타입을 사용. dags 폴...

https://velog.io/@bellship24/EKS-위에-Airflow-v2.5.3-Helm-배포하기

Kubernetes환경에서 Airflow를?!?!

https://developnote-blog.tistory.com/176

2. Airflow on Kubernetes - git sync

Prerequisite 1. Airflow on Kubernetes Apache Airflow는 쿠버네티스에 친숙한 프로젝트를 목표로 하고 있으며, 공식 helm 차트를 관리하고 있다. 해당 차트를 이용해서 쿠버네티스 클러스터에 비교적 쉽게 에어플로우를 구축할 수 있다. G jinyes-tistory.tistory.com Native Airflow를 사용하면 local directory에 dag를 저장하고 사용하면 되지만, kubernetes는 컨테이너 형태(상태)로 리소스를 관리하기 때문에, 언제든 Airflow가 종료되었다가 재시작이 될 수 있다. Airflow는 remote directory 기능인 git sync를 제공하는데, 해당 기능을 이용해서 DAG 파일 저장소로 github reposito..

https://kimjinung.tistory.com/m/5

EKS 위에 Airflow 구성

EKS 생성은 모두 끝났다고 가정하고 진행합니다. 이 글은 Kubernetes를 설명하는 글이 아니므로 namespace, pod, kubectl, helm 등 K8S와 함께 등장하는 용어에 대한 깊이 있는 설명은 하지 않습니다. 로컬 PC에서 aws 명령어를 통해 EKS 컨텍스트를 추가합니다. 무슨 의미인지 해석하실 필요 없습니다. 쉽게 생각해서 로컬에서 kubectl 명령을 사용할 때 바라보게 되는 곳이 AWS EKS라고 생각하시면 됩니다[1]. --name 뒤에 transformer-dev는 각자의 설정에 맞게 변경해주시기 바랍니다. transformer-dev는 이 글이 작성될 때 기준으로 테스트한 EKS 클러스터 이름입니다. $ aws eks --region ap-northeast-2 updat..

https://jybaek.tistory.com/956

Apache Airflow와 Amazon EKS가 만났을 때 벌어지는 일

그런데 이제 GPU와 Neuron을 곁들인

https://medium.com/musinsa-tech/sre-9fe5113de898

24.03.20

airflow 무한 pending 상태

[Airflow] Helm을 사용하여 k8s cluster에 Airflow 설치

Airflow Chart 설치 시 배포 되는 PVC에 연결될 PV를 미리 배포 해줍니다. 10GB 2개, 110GB 2개로 설정해주었습니다. 최소 용량이 얼마나 필요한지 모르겠으나 적어도 2개 PVC가 110GB 이상을 요구하는 것 같았습니다. 그렇지 않은 경우 Pending 상태로 무한 대기상태에 빠지더군요.

https://ysbaekfox.github.io/airflow/airflow-helm-install/

Airflow on Kubernetes with Helm

In the cloud-native world, Kubernetes and Helm are becoming the de-facto standards for deploying, managing, and scaling applications. As a…

https://hungngph.medium.com/airflow-on-kubernetes-with-helm-c795545325dc

Error in deploying airflow by kubectl+kubeadm+helm+caiico cni

Description I installed airflow from offical kubectl, kubeadm, helm, calico cni website but airflow can't be deployed. Tried Install kubectl, kubeadm, helm from offical website. Install calico cni...

https://devops.stackexchange.com/questions/18995/error-in-deploying-airflow-by-kubectlkubeadmhelmcaiico-cni

나와 동일한 에러를 발견

8560

issues

1795

rancher storage class

•

storageClass는 동적으로 PersistentVolume을 생성

•

사용자가 PVC를 생성하고, 이 PVC가 특정 StorageClass를 참조

•

아래 storage class를 참고

https://github.com/rancher/local-path-provisioner

•

PVC는 사용자가 요청하는 스토리지 리소스를 정의