데이터 엔지니어링 (w/ Airflow, k8s)

date

Sep 4, 2023

thumbnail

slug

airflow

author

status

Published

데이터 엔지니어링이란?

근래에 데이터 관련 직군에서 엔지니어링 업무쪽에 전문적으로 오너십을 갖는 데이터 엔지니어라는 세부적인 직무가 생겨나고 있습니다

데이터의 ETL(Extract, Transform, Load)가 중요해지는 만큼 해당 직무는 데이터를 다루고 data driven하게 업무를 하는 회사에서는 Data Engineering은 간과해서는 안 될 중요한 부분입니다.

실제 서비스를 제공하는 IT 회사에서는 앱 또는 웹으로 사용자에게 서비스를 제공하면서 로그 데이터, 트렌젝션 데이터, 서비스 데이터 등 다양한 데이터가 생성되게 됩니다.

이러한 데이터를 수집해서 데이터 저장소(RDB, Storage 등)에 저장하게 됩니다. 이러한 데이터를 바탕으로 데이터 웨어하우스, 데이터 레이크를 만들고 이를 새로운 형태로 변환하고 분석하고 EDA되는 과정, 다른 플랫폼과의 데이터 전송 등의 모든 업무가 데이터 엔지니어링 입니다. 회사에 따라서 데이터 엔지니어링과 데이터 분석(DA)을 세분화하여 나눌 수 있고, 이를 하나의 DE팀이 담당하여 하는 경우도 있습니다.

이런 과정 속에서 ETL, 리소스, 네트워크 및 인프라 관리, 새로운 데이터 플랫폼 개발, SQL 스킬 및 쿼리 튜닝, MLOps 등 다양한 도메인 지식이 필요하고 이를 전반적으로 공부하기 때문에 개인적으로 백엔드 개발과 데이터 양쪽에 걸쳐있는 직무라고 생각합니다.

이러한 데이터들이 새로운 비즈니스 모델을 제시하거나 의사결정 등을 할때 중요한 지표의 역할이 될 수 있고, 이러한 데이터들을 활용하여 ML, DL, AI의 학습 데이터, 모델링에 활용할 수 있습니다.

💡

데이터 엔지니어가 하는 일에 대해 자세한 내용을 원하시면, 아래 카일님이 잘 정리를 해 주신 영상을 보시면 도움이 됩니다. https://www.youtube.com/watch?v=1Viu7Ce632s

이러한 데이터 엔지니어링을 조금 쉽게 할 수 있도록 도와주는 도구가 바로 Airflow 입니다.

Airflow 란?

워크플로우를 코드로 작성할 수 있고 스케줄링하고 모니터링 하는 플랫폼

Airflow의 구성요소

DAG

개발자가 코드로 작성하는 워크플로우(흐름)

Task

하나의 작업 단위

예) 서비스 DB의 데이터를 데이터 웨어하우스로 가져온다.

Batch

특정 트리거 또는 시간에 따라 반복적으로 작업을 수행할 수 있습니다. (Cron Job)

Real Time

실시간으로 실행합니다. (e.g. Log monitoring 등)

Airflow의 동작 원리

출처 : https://engineering.linecorp.com/ko/blog/data-engineering-with-airflow-k8s-1/

Web Server

Web UI에 접속하기 위한 서버

Scheduler

개발자가 정의한 Python dag 스크립트를 주기적으로 읽습니다.

개발자가 정의 해놓은 특정 시점에 dag를 실행합니다.

Task는 Dag내 어떤 형태로 Workflow가 구성되어 있는 지를 meta DB에 기입을 합니다.

Executor

현재 시점에 실행해야하는 Task를 주기적으로 찾습니다.

실행해야하는 Task가 있다면, 상태를 변경하고 내부 Queue에 Task를 넘겨주게 됩니다.

Worker

전달받은 Task를 실행하게 됩니다.

실행하는 Task의 형태는 CLI Command 입니다.

다시 말해, CLI Command가 워커에서 실행되는 구조입니다.

Executor의 종류

Executors는 크게 Local Executors와 Remote Executors 로 구분할 수 있습니다.

Local Executors

Debug Executor

Local Executor

Sequential Executor

Remote Executors

Celery Executor

CeleryKubernetes Executor

Dask Executor

Kubernetes Executor

LocalKubernetes Executor

Executor는 Task 인스턴스를 실행하는 주체입니다. 종류에 따라, Airflow가 다르게 동작을 합니다.

Airflow Executor에는 Base Executor의 부모 클래스를 상속 받는 여러 Executor가 있습니다.

현업에서 가장 많이 사용되는 것은 Local Executor와 Celery Executor입니다.

LocalExecutor

LocalExecutor는 airflow.cfg 에 있는 parallelism factor의 값의 여부에 따라 두 가지로 나눠지게 됩니다.

_UnlimitedParallelism

해당 값이 0이면 이론 상 Task를 무한히 생성할 수 있습니다.
worker의 구현체는 Local Worker가 되게 됩니다.
result_queue 하나만 갖습니다.

_LimitedParallelism

1 이상 경우는 Task는 유한하게 생성할 수 있습니다.
worker의 구현체는 QueuedLocalWorker가 되게 됩니다.
task_queue, result_queue 두개를 갖습니다.

고가용성에 문제가 있어, beta 환경이나 dev 환경에서 사용합니다.

CeleryExecutor

내부적으로 queue를 가지고 있는것은 동일

Task를 외부 message broker을 통해 비동기로 전송

전송된 Task를 Worker Machine의 Celery Worker가 가져가서 CLI Command로 실행

KubernetesExecutor

다양항 형태의 소스 데이터 환경

다양한 형태의 타겟 데이터 환경

각각의 서비스에서 만들어지는 데이터의 형태와 환경이 각각이고 다른 경우

MySQL, MSSQL, NoSQL, byte, 클라우드, 하드 클러스터 등

Airflow on Kubernetes vs Kubernetes Executor

Airflow on Kubernetes

일반적인 Airflow on Kubernetes를 통해 로컬 클러스터를 구성한다면,

단순히 웹서버와 스케줄러

Web server와 Scheduler를 하나의 Pod

Celery를 포함한 Airflow

Web server, Scheduler Pod

Redis or RabbitMq 등 메시지 브로커 cluster를 포함한 Pod

Airflow Worker Pod

장점

구성이 간단

Template 화 하기 쉽다

단점

유지보수 비용

의존성에 따라 무거워 짐

Kubernetes의 장점을 살리지 못함.

이러한 단점을 극복하기 위해 KubernetesExecuter를 사용

Kubernetes Executer

필요할 때만 Kubernetes 자원을 사용

필요한 Docker Container만 골라 Pod로 실행

K8sPodOperator

일반적인 Operator

일반적인 BashOperator, PythonOperator은 task가 들어오면 k8sExecutor가 airflow.cfg에 kubernetes session에 관리자가 정의한 k8s config와 이미지 정보를 가지고 worker를 만듭니다.

Airflow Worker Pod를 생성한 후, 이 Worker Pod가 일반 Operator를 실행하게 되고, 작업이 완료되면 리소스를 반납하고 종료되게 됩니다.

K8sPodOperator

Airflow Worker Pod를 생성하는 것 까지는 기본적인 Operator와 동일합니다.

Worker에서 개발자가 dag script에 직접 정의한 custom Pod가 수행됩니다.

다시말해, worker에서 내가 실행하고 싶은 task를 또 다른 Pod로 실행합니다.

Kubernetes

일반적으로 쿠버네티스의 경우는 DevOps나 백엔드에서 마이크로서비스 아키텍쳐를 구성할 때 많이 사용합니다.

하지만 데이터 엔지니어링에서도 쿠버네티스는 스케줄러, 태스크의 리소스 관리, 오퍼레이션을 유용하고 쉽게 할 수 있습니다.