[MLOps] Kubeflow pipeline handling (1) : pipeline 개념 및 기본 진행 방법

Kubeflow Pipeline

Kubeflow pipeline이란, ML workflow를 생성할 수 있는 kubeflow compoenent를 말한다.

Pipeline은 아래 그림처럼 workflow의 각각의 component들과 dependency를 포함하여 DAG 형식을 나타내고 있다.

 

즉, 파이프라인은 Component라는 단위로 쪼개지며, docker image로 packaging 되어 실행

--> 데이터가 component에 입력되어 결과를 출력하는 방식

 

그래서 pipeline 작업 영역은 experiment라는 단위로 논리적으로 분리시킴

--> 즉, experiment는 pipeline이 실행되는 workspace를 의미

 

그리고 각 pipeline은 run이라는 단위로 관리할 수 있다.

이 컨테이너화된 아키텍처는 워크플로우가 변경될 때 컴포넌트들을 간단히 재사용, 공유, 교체할 수 있게한다 - https://litiblue.com/post/kubeflow-pipelines/

구성 요소
  1. UI : Experiment, Job 또는 Run을 관리하고 추적
  2. Engine : 단계별 ML workflow를 예약
  3. SDK: Pipelines 및 components를 정의하고 조작
  4. (Jupyter) Notebook : SDK를 사용하여 시스템과 상호작용

 즉, Notebook에서 SDK를 이용해 pipeline을 작성하여 업로드하면, UI를 통해 확인할 수 있는 구조

 

그래서 Kubeflow pipeline의 목표는

 1. ML / DL training pipeline의 orchestration을 단순화

 2. 다양하게 시도된 trial과 experiments를 관리

 3. component와 pipeline을 재사용

 --> 다시 build할 필요 없는 end-to-end solution을 생성하는 것이다.

 

---

Kubeflow pipeline engine

Kubeflow는 Argo workflow engine을 사용하여  pipeline을 생성

즉, k8s pod가 시작 -> pod가 container 시작 -> container가 program을 실행하는 구조

 

Argo와 Airflow의 차이점이 궁금하다면, 다음 글을 정독

Argo vs Airflow: Which is Better for Your business | Hevo

 

---

Pipeline Component

데이터 전처리, 데이터 변환, 모델 학습 등과 같은 ML workflow에서 1 step을 수행하는 코드 집합을 의미

 

 - 이름, 매개변수, 반환값이 있다는 점에서 function과 유사

 - python function을 통해 생성 가능

 - 이미 만들어진 컴포넌트가 container image로 만들어져 있다면, 이미지를 별도로 호출해서 사용 가능

 

구성 요소

 - Metadata : name, description, etc.
 - Interface : input / output 내용 (name, type, default value, etc)
 - Implementation : component 실행 시 입력되어야 할 input argument value와 conainer image를 정의

 

---

Kubeflow Pipeline Example

1. python function로부터 component 생성

kfp 라이브러리는 1.8.14 버전으로 설치해서 진행 예정

# jupyter notebook 파일 내에서 진행

#kfp 1.8.14 버전 설치
!pip install kfp==1.8.14

# 함수로부터 컴포넌트를 만들기 위해 메서드 import
from kfp.components import create_component_from_func

# 함수 description 확인
create_component_from_func??

notebook에서 code 실행 시 나타나는 화면

 

설명을 읽어보면, custom 함수 정의 후 create_component_from_func의 argument로 입력 시, operator가 생성된다고 한다.

component를 정의 후 decorator를 붙여주면, operator가 생성됨을 확인할 수 있다.

 

이후 이 component를 yaml 형식으로 저장해보자

# notebook cell에서 코드 실행
component_name = "0-1_hello-world-component.yaml"
hello_world_component.component_spec.save(component_name)

 

해당 cell 실행 시, root directory에 yaml 파일이 저장되어 있음을 확인할 수 있다.

 

앞서 언급했듯이 name, output, implementation 3가지가 들어있는 걸 확인할 수 있다.

 

implementation을 잠깐 살펴보면,

 - container image가 python 3.7이라고 되어 있다.

 - 컨테이너 내부에서 실행될 command가 지정이 되어 있다.

 - 우리가 입력한 hello_world_component()가 입력되어 있는 걸 확인할 수 있다.

 

2. component 생성 후 pipeline 업로드

우리가 만든 component를 pipeline에 입력하기 위해서는 dsl의 pipeline이라는 함수를 정의하고 decorate해주어야 한다.

# kubeflow pipeline 라이브러리 import
import kfp

# custom pipeline 정의 후 kfp로 decorate 진행
@kfp.dsl.pipeline(name="hello_world_pipeline1", 
                  description="Hello World Pipeline!")
def hello_world_pipeline():
    # 보통 pipeline 생성 시 component function이 있으면 task라고 이름을 지정해줌
    hello_world_task = hello_world_component()

# 이후 complie 시 pipeline file이 나오는 걸 확인할 수 있음
kfp.compiler.Compiler().compile(hello_world_pipeline, "hello-world-pipeline.zip")

 

pipeline file 생성 완료했으면, 다음 과정을 따른다.

 

 

위와 같이 파이프라인 파일을 다운로드 받고, kubeflow 페이지에서 pipeline upload 버튼을 누르고 값을 기입 후 파일을 upload 후 create 버튼을 누르면 pipeline이 생성된다.

 

해당 과정을 완료하면, pipeline이 업로드가 된 걸 확인할 수 있다.

pipeline이 upload가 된 모습

 

3. Pipeline 실행

앞서 말했듯이, Pipeline을 실행하기 위해서는 experiment가 생성이 되어야 한다.

 

해당 페이지 우측 상단에, [Create experiment] 라는 버튼을 누른다.

experiment  세부 설정 내용

이후 이름과 설명만 기입하면 experiment가 생성된다.

 

Experiment 생성 시 run 옵션을 지정할 수 있다.

다시 말하지만 Run은 pipeline의 single execution을 의미한다.

Run의 각 옵션을 살펴보면,

 - Experiment : experiment를 지정 가능. 우리는 lab0만 생성했기 때문에 할당이 자동으로 lab0로 된 것이고
 - RunType
     - One-off : 한 번만 실행하는 것
     - Recurring : pipeline을 반복해서 주기적으로 실행하는 것 / run-trigger에 의해 반복됨 / periodic or cron으로 설정 가능

 

옵션을 지정 후 [Start] 를 누르면, pipeline이 시작되는 걸 확인할 수 있다.