Kubeflow를 이용한 Vertex Pipeline 만들기 (Intro)

TFX를 사용해서 ml pipeline을 만드는 법에 대한 피드백을 저번 글에서 했었는데, 이번에는 Kubeflow를 사용해서 ml pipeline 을 만드는 방법을 공유하고자 합니다. 약간의 삽질(?)이 필요하지만, 꾸준히 하다보면 감을 잡으실 수 있을거라고 생각합니다. 오늘 소개할 내용은 codelabs.developers.google 에서 제공하는 tutorial을 따라가는 내용이고, 그 전에 gcp에 대한 경험이 좀 있으면 좋을 것이라고 생각합니다. 

 

일단 Vertex Pipeline을 실행(?)이라고 해야되나 만든다고 해야되나.. 하기 위해서는 여러가지 방법이 있습니다. 그 중 하나는 google console에 들어가서 vertex pipeline에 들어가서 코드를 올리는 방법이 있고, 또 jupyter notebook에서 kfp를 사용해서 돌리는 방법이 있습니다. 하지만 저는 그 두 가지 방법을 사용하지 않고, local 에서 Vertex AI SDK를 사용하는 방법을 활용하였습니다. 이를 위해서는 해야되는 일들이 몇 가지 있습니다.

 

일단 GCP Project가 있어야 하고, 그 Project에서 pipeline을 만들고 돌릴 수 있을 만한 권한이 갖춰진 Service account가 있어야 합니다. 또, Local에서 Vertex AI SDK를 사용하기 위해서는 Google cloud SDK를 설치해야됩니다. gcloud init을 커맨드 입력해서 연결을 한 이후에, 상황에 따라서 다시금 authentification을 해야될 수도 있습니다. gcloud auth login 을 통해서 가능합니다. 또한, kfp나 google-cloud-aiplatform 를 pip install 해야될수도 있습니다. (venv 권장)

 

그럼 이제 본격적으로 첫번째 파이프라인을 만들어볼까요?

import kfp

from kfp.v2 import compiler, dsl
from kfp.v2.dsl import component, pipeline, Artifact, ClassificationMetrics, Input, Output, Model, Metrics

from google.cloud import aiplatform
from typing import NamedTuple

관련 라이브러리들을 먼저 import 한 이후에,

@component(base_image="python:3.9", output_component_file="first-component.yaml")
def product_name(text: str) -> str:
    return text

 

pipeline은 각각의 component로 이뤄져 있고, component는 kfp에 있는 module중 하나입니다. component를 decorater로 부르면서 base_image는 container image를 불러오고,  output_component_file은 optional이라고 합니다. 끝나고 나니깐 first-component.yaml 파일이 생기는 걸 볼 수가 있습니다. 이 함수는 그저 string이 들어오면 string이 나오는 function을 말하네요.

 

@component(packages_to_install=["emoji"])
def emoji(
    text: str,
) -> NamedTuple(
    "Outputs",
    [
        ("emoji_text", str),  # Return parameters
        ("emoji", str),
    ],
):
    import emoji

    emoji_text = text
    emoji_str = emoji.emojize(':' + emoji_text + ':', language='alias')
    print("output one: {}; output_two: {}".format(emoji_text, emoji_str))
    return (emoji_text, emoji_str)

 

두 번째 component 입니다. input 이 text가 들어오면, 이를 emojize해서 return 하는 함수네요.

 

@component
def build_sentence(
    product: str,
    emoji: str,
    emojitext: str
) -> str:
    print("We completed the pipeline, hooray!")
    end_str = product + " is "
    if len(emoji) > 0:
        end_str += emoji
    else:
        end_str += emojitext
    return(end_str)

 

마지막 component입니다. 세 개의 input을 담아서, emoji 나 emojitext 를 붙여 주네요. 이제 어떻게 앞의 두 개의 component들에서 build_sentence에 output을 넣어줄까요?

@pipeline(
    name="hello-world",
    description="An intro pipeline",
    pipeline_root=PIPELINE_ROOT,
)

# You can change the `text` and `emoji_str` parameters here to update the pipeline output
def intro_pipeline(text: str = "Vertex Pipelines", emoji_str: str = "sparkles"):
    product_task = product_name(text)
    emoji_task = emoji(emoji_str)
    consumer_task = build_sentence(
        product_task.output,
        emoji_task.outputs["emoji"],
        emoji_task.outputs["emoji_text"],
    )

 

다음과 같이 pipeline decorator를 사용해서 pipeline을 만들어줍니다. PIPELINE_ROOT 는 GCP에서 하는 경우에는 GCP Storage Bucket을 활용할 수 있겠죠? 위의 intro_pipeline 함수에서 각각의 component들의 연결고리를 만들어주는 겁니다. 어떤 component 의 output이 어디로 들어가는지에 대해서.. 

 

compiler.Compiler().compile(
    pipeline_func=intro_pipeline, package_path="intro_pipeline_job.json"
)

그리고 intro_pipeline을 compile해주는 일을 합니다. 

 

from datetime import datetime

TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")

job = aiplatform.PipelineJob(
    display_name="hello-world-pipeline",
    template_path="intro_pipeline_job.json",
    job_id="hello-world-pipeline-{0}".format(TIMESTAMP),
    enable_caching=True
)

job.submit()

 

마지막으로, pipelineJob을 정의해주고 submit하면 끝이납니다. 이 파일을 실행시켜주면, 

 

Reference 1 : https://cloud.google.com/vertex-ai/docs/pipelines/build-pipeline 

Reference 2 : https://codelabs.developers.google.com/vertex-pipelines-intro#4