반갑습니다.
AirFlow에 대해 알아볼까 합니다.
저번 글에서 Serving에 대해 다뤘고, 이번에는 좀 더 깊게 들어가 Batch Serving에 활용되는 Apache Airflow에 대해 소개해보겠습니다.
Batch Serving?
Batch Serving이란, ML 코드를 주기적으로 실행해서 결과를 특정 시간마다 제공하는 Serving을 의미합니다.
이전 글에서도 작성했지만, 한번 더 짚고 넘어갑시다
ML/AI 도메인에서 Batch Processing을 수행하는 것이 Batch Serving입니다.
예측 코드를 정한 시간마다 실행시켜 결과를 제공하는 형태로 구현된다고 보시면 되겠습니다.
데이터베이스/웨어하우스에 존재하는 데이터를 추출한 뒤, 일괄로 예측하는 방법이 있을 수 있겠죠.
Crontab
Linux에 존재하는 기능입니다.
OS에 의해 정해진 시간에 맞게 프로그램을 실행하도록 할 수 있습니다.
Cron 표현식을 사용하여 Batch processing에 대한 시간을 정할 수 있습니다.
위 사진이 Cron 표현식입니다.
위에 맞는 규칙에 따라 작성된 크론 표현식의 의미들입니다.
예시를 보는게 확실히 이해하기 쉽군여 ㅎㅎ
Batch processing을 Crontab을 통해 수행할 수 있지만, Crontab은 몇가지 단점이 존재합니다.
- 재실행 알림이 안된다.
파일을 실행하다가 오류가 발생하면, 별도의 처리가 없습니다.
한번만 실행되기에 두세번 더 Try 하도록 설정할 수도 없지요. - 과거 실행 이력 및 로그를 보기가 어렵다.
- 여러 파일을 실행하거나, 복잡한 파이프라인을 만들기 힘들다.
분기점이 없고 일직선 방향으로 코드가 실행되면 문제가 없겠지만, A상황 B상황에 따라 다른 코드가 실행되도록 하고 싶다면 Crontab으로는 어려움이 많습니다.
이러한 단점을 보완하기 위해 더 정교한 스케줄링 및 워크플로우 도구의 필요성이 인지되었고, argo, Apache Airflow 등 다양한 도구들이 개발됐습니다.
이번 글에서는 그 중 Airflow에 대해 소개합니다.
Airflow
Airflow는 Airbnb에서 개발된 스케줄링 도구입니다.
업데이트가 빠르며, 스케줄링 도구로서 무겁지만 거의 모든 기능을 제공하고 확장성이 좋아 일반적으로 스케줄링과 파이프라인 작성 도구로 많이 사용됩니다.
Airflow 등장 후, 스케줄링 및 워크플로우 도구의 표준이 됐다고 합니다.
그만큼 잘 만든 툴이라 그렇겠죠.
Airflow를 사용하는 이유
Airflow를 사용하는 이유는 크게 세 가지로 볼 수 있습니다.
- 워크플로우를 관리할 있음
- 데이터 파이프라인 흐름을 스케줄링하고 모니터링할 수 있음
- 데이터 처리 파이프라인을 효율적으로 관리할 수 있음
Airflow는 어떻게 사용할까요
위 사진처럼 DAG이라는 것을 만들고, Task들을 지정합니다.
그 뒤 지정한 Task들을 알맞게 편성해주면 하나의 DAG이 됩니다.
t1 이후 t2를 수행하고 싶다면, t1 >> t2 와 같은 방식으로 편성합니다.
이후 웹 서버를 실행하면 위와 같은 페이지가 나옵니다.
DAGs에서 수행할 Task를 On Off로 수행하거나, Cron 표현식으로 정해진 시간마다 자동으로 수행하도록 합니다.
그 과정을 상세히 보실 수 있습니다.
위처럼 DAG 하나하나 안에 구성 Task들이 성공했는지 실패했는지 등 다양한 상태를 확인합니다.
또, DAG 선언 시에 Retries를 통해 Task가 실패될 경우 몇번 더 반복할지에 대해서도 설정할 수 있습니다.
Airflow의 핵심 개념
-
DAGs(Directed Acyclic Graphs)
순환하지 않는 방향이 존재하는 그래프를 의미하는 DAG은 Airflow에서 작업을 정의하는 방법입니다. (Airflow에서만 사용되는 용어는 아닙니다.)
작업의 흐름과 순서를 정의합니다.
$AIRFLOW_HOME/dags로 정의한 폴더 내에 .py파일을 넣으면 모두 탐색되고 DAG이 파싱됩니다. -
Operator
Airflow의 작업 유형을 나타내는 클래스입니다.
BashOperator, PythonOperator, SQLOperator 등 다양한 Operator가 존재합니다. -
Scheduler
Scheduler는 각종 메타 정보의 기록을 담당합니다.
DAG Directory 내 .py 파일에서 DAG을 파싱하여 DB에 저장합니다.
DAG들의 스케줄링을 관리합니다.
실행 진행 상황과 결과를 DB에 저장합니다.
Executor을 통해 스케줄링 기간이 된 DAG을 실행합니다. -
Executor
Executor는 스케줄링 기간이 된 DAG을 실행하는 객체입니다.
크게 Local Executor, Remote Executor로 나뉩니다.
Executor에는 LocalExecutor, CeleryExecutor, Kubernetes 등 다양한 Executor가 존재합니다.
주로 규모에 따라 redis, Kubernets와 같이 사용합니다.
Airflow의 기본 아키텍처는 위와 같은 구조입니다.
그림으로 보면 좀 더 와닿을듯 합니다.
DAG 작성
with DAG(
dag_id = 'test1',
description = 'it is test dag',
start_date = dats_ago(2), # 언제부터 시작할지 정의합니다.
schedule_interval = "0 6 * * *", # 크론 표현식을 통해 언제 실행될지 정의합니다.
tags = ['my_dags'], # 태그를 설정합니다.
) as dag:
task1 = BashOperator(
task_id = 'task1',
bash_command = 'echo itis',
owner = 'Kimyongari',
retries = 3, # 오류가 발생할 경우 다시 실행할 횟수
retry_delay = timedelta(minutes = 5), # 다시 실행할 때의 텀
)
task2 = PythonOperator(
task_id = 'task2',
python_callable = print_test, # 함수를 넣는 형식
depends_on_past = True, # 이전의 Task가 제대로 동작해야 실행
owner = 'Kimyongari',
retries = 3,
retry_delay = timedelta(minutes = 5)
task1 >> task2위와 같이 작성하신 뒤 .py 파일을 dags폴더에 넣으면 Web에서 확인할 수 있습니다.
'
'
'
Airflow에 대해 글을 작성해봤습니다.
사실 현업에서는 클라우드 서비스나 VM, Kuebernetes들과 엮어서 사용하는데...
배포에 대해 잘 알지 못하니 더 깊게 설명을 잘 못하는 부분이 죄송스럽습니다.
제가 잘알이 된다면 Kubernetes나 Docker Compose에 대해 설명하면서 추가로 내용을 작성해보겠습니다.
감사합니다 !
