ML Pipeline Solution

머신러닝 디자인 패턴 책을 공부하면서 여러 Machine Learing Pipeline을 소개해 정리해 보았다.

TFX(TensorFlow Extended)

TFX | ML 프로덕션 파이프라인  |  TensorFlow

구글에서 만든 플랫폼으로 이름에서도 알 수 있듯 TensorFlow에 기반하고 있다. ML 시스템을 정의하고, 실행하고, 모니터 하는데 필요한 component들을 통합하는 라이브러리이다. Orchestration은 Airflow나 Kubeflow 같은 다른 플랫폼을 사용하여 할 수도 있다(Orchestration이란 파이프라인에 로직을 추가하여 실행할 단계와 그 결과를 결정하는 것이다). 각 컴포넌트 라이브러리는 아래와 같은 흐름을 가지고 이용된다고 한다.

 

출처: https://www.tensorflow.org/tfx/guide

• 데이터 수집
  • ExampleGen: 여기서 example이란 ML에서는 학습에 사용되는 라벨이 지정된 인스턴스를 의미한다. 따라서 데이터를 생성하는 컴포넌트로 외부에서 데이터를 가져오는 역할을 한다. 이 컴포넌트를 사용하면 csv, TFRecord, 빅쿼리, 커스텀 소스에서 데이터를 가져올 수 있다.
• 데이터 검증
  • StatisticsGen: 수집된 데이터의 요약 통계를 생성한다.
  • SchemaGen: 통계를 검증하고, schema를 출력한다.
  • ExampleValidator: 이상치(anomalies)와 missin value를 탐지한다. 데이터 드리프트의 징후나 잠재적 학습 편향도 확인한다.
• 데이터 전처리
  • Transform: feature engineering을 하는 컴포넌트이다. feature engineering이란 데이터를 모델에 적합한 경식으로 변환시키는 것이라고 생각하면 된다. 임베딩, 정규화 등이 포함될 수 있다.
• 학습
  • Trainer: 이름 그대로 모델학습에 사용된다.
• 배포
  • Pusher: 서빙 인프라에 모델을 배포한다. 

 

Kubeflow Pipeline

Introduction

Kubeflow pipeline은 ML pipeline에 존재하는 다양한 일들을 조합(orchestration)하는 일을 단순화시켜주는 플랫폼이다. 쉽게 실험할 수 있고, 재사용할 수 있게 도와준다. 실험과 실행들을 관리할 수 있는 UI, workflow를 스케줄링하는 엔진, pipeline과 component를 정의하고 다룰 수 있는 SDK를 제공한다.

 

 

Apache Airflow

What is Airflow? — Airflow Documentation

Airflow는 batch기반 workflow의 개발, 스케줄링, 모니터링을 위한 오픈소스 플랫폼이다(ML에 특화되었다기보다는 범용적이라고 한다). 웹 인터페이스를 통해 workflow의 상태를 관리할 수 있다. 또한, 랩탑부터 분산 설정까지 다양한 방법으로 배포가능하다. Airflow는 DAG(Directed Acyclic Graph, 단방향 비순환성 그래프)를 통해 pipeline을 생성한다. DAG는 Python을 통해 작성가능하다. 아래는 document의 예시 코드이다. DAG는 ' >>', '<<'로 표현한다고 한다.

from datetime import datetime

from airflow import DAG
from airflow.decorators import task
from airflow.operators.bash import BashOperator

# A DAG represents a workflow, a collection of tasks
with DAG(dag_id="demo", start_date=datetime(2022, 1, 1), schedule="0 0 * * *") as dag:

    # Tasks are represented as operators
    hello = BashOperator(task_id="hello", bash_command="echo hello")

    @task()
    def airflow():
        print("airflow")

    # Set dependencies between tasks
    hello >> airflow()

아래 이미지는 airflow architecture를 간단히 나타낸 것이다. 각 구성요소를 간단히 알아보자. 시작은 scheduler부터.

출처:&nbsp;https://airflow.apache.org/docs/apache-airflow/stable/concepts/overview.html

• scheduler: 스케줄 된 workflow의 triggering, 모니터링과 executor에 시작할 태스크를 할당한다
• executor: 실행 중인 태스크를 다룬다. default airflow에서는 scheduler 안에서 모든 것을 실행하지만, 대부분에 prouction에 적합한 executor들은 worker에서 실행한다. executor 종류에는 Celery Executor, Kubernetes Executor 등이 있다.
• webserver: UI 제공
• DAG directory: DAG가 존재하는 폴더로 scheduler, executor, worker에서 이곳에 있는 파일을 읽는다
• metadata database: scheduler, executor, webserver에서 사용하는 것으로 상태(state)를 저장한다.

+) 오늘의 집에서 airflow 도입 과정에 대해 글을 쓴 것이 있는데 airflow를 이해하는데 좋은 것 같다. 여기

MLflow

MLflow는 크게 Tracking, Projects, Models, Model Registry로 나누어져 있다. 각 component는 ML workflow를 관리하는 데 있어 도움을 준다.

MLflow Tracking은 parameter, code version, metrics, artifact 등을 로깅하는 API와 UI이다. Project는 code를 재사용가능하게 패키징 하는 표준적인 포맷을 제공한다. 예를 들면, yaml파일을 통해 Python환경을 동일하게 유지할 수 있도록 도와주는 것이다. Models는 조금 더 배포에 관련되어 있다. 다양한 방식(flavor)으로 ML model을 패키징 하기 위한 convention을 제공한다. Model Registry는 이름 그대로 모델 저장소라고 생각하면 된다. 각 component를 살펴보았을 때 앞서 보았던 플랫폼들의 기능과는 조금 달라 보인다는 생각이 들었다. 알고 보니 MLflow pipeline component가 따로 존재했다.

 

MLflow Pipelines (experimental) — MLflow 1.28.0 documentation

MLflow Pipelines는 실험적으로 운영되고 있는 것 같다. MLflow에서 Pipeline은 repository단위로 관리되며 그 안에 steps 디렉토리에 필요한 단계의 코드들을 넣고 yaml을 통해 Pipeline을 설계한다.