💡

[VertexAI Pipeline]#1基礎編 Hello World！

2022/05/05に公開

Google Cloud

こんにちは！ chameleonです！

最近はマネージドサービスがかなり充実してきて、Kubeflow PipelineもVertexAIのPipelineサービスでアプリケーションレイヤーだけプログラミングすればインフラレイヤーを気にせずに実行できるようになってきましたね。

今後はGCP上で機械学習系のワークフローを組む際はVertexAIのPipelineが利用されるシーンが多くなってくるかと思いますので、VertexAI pipelineを利用できるようになっておくとエンジニアとしての市場価値も高くなってくるかと思います。

学習を始める際にはミニマム構成で基本を確認することから始めると学習がしやすくなります。
今回は１つのコンポーネントでhello-worldという文字列をログに出力するワークフローを作成して、実行するところから初めてみましょう。

想定する読者

・機械学習系のワークフローを作ってみたい
・KubeflowPipelineを使ってワークフローを構築てみたい
・VertexAI Pipelineを使ってみたい

ゴール

今回はVertexAIのPipelineサービスを利用して、「hello-world」をログに出力するワークフローを作成していきます。下記のよう感じでPipelineサービスのUIから作成されたワークフローとログが確認できます。

構築の大まかな流れ

0.必要なライブラリの宣言
1.処理を行うコンポーネント(関数)の作成
2.パイプラインの作成
3.コンパイルする対象のパイプラインと出力json名を指定してコンパイルを実行
4.コンパイルしたパイプラインの実行

構築スタート!!!

0.必要なライブラリの宣言

まずはパイライン作成に必要なライブラリのインポート宣言を行います。

from kfp.v2.dsl import component #コンポーネント作成用
from kfp import dsl #パイプライン作成用
from kfp.v2 import compiler #コンパイラ用
import google.cloud.aiplatform as aip

1.処理を行うコンポーネント(関数)の作成

まずは、処理を行うコンポーネント(関数）を作成していきます。
@componentというデコレータを利用してメソッド定義を行います。

@component(base_image="python:3.7")
def say_message(message:str):
    print(f"{message}")

今回は引数で指定した文字列を出力するメソッドをコンポーネントとして定義します。

2.パイプラインの作成

次はメインとなるパイプライン（ワークフロー）を定義します。
パイプラインの定義は@dsl.pipelineというデコレータを利用します。

# パイプラインの作成
# 手順2:パイプラインの作成
@dsl.pipeline(
    name="tutorial-pipeline1"
)
def tutorial_pipeline():
    say_message(message="hello-world")
    pass

デコレータの引数にはパイプラインの名前（好きな名前）を指定します。
手順1で作成したsay_messageコンポーネントを呼出し、引数のmessageに「hello-world」を指定して呼び出します。

ここまでで、パイプライン作成は完了です。

3.コンパイルする対象のパイプラインと出力json名を指定してコンパイルを実行

上記で作ったパイプラインをコンパイルします。
手順2で作成したpipeline関数とコンパイル結果として出力するjsonファイルの出力先を指定します。

# コンパイル
# 手順3:コンパイルする対象のパイプラインと出力json名を指定してコンパイルを実行
compiler.Compiler().compile(
    pipeline_func = tutorial_pipeline,
    package_path  = "/home/npprw160/Sandbox/kpf/tuto_pipeline_v1.json",
)

4.コンパイルしたパイプラインの実行

最後にコンパイルしたjsonファイルを実行してVertexAIのpipelineサービスでワークフローを実行します。
aip.initでGCPプロジェクトのIDと処理を行いロケーションを指定して初期化を行います。

最後にパイプラインの表示名と、手順3でコンパイルしたjsonファイルのパスを指定して実行します。

# パイプラインの実行
# 手順4:コンパイルしたパイプラインの実行
aip.init(
    project="awesome-height-240603",
    location="us-central1"
)

aip.PipelineJob(
    display_name = "Pipeline基礎編",
    template_path = "/home/npprw160/Sandbox/kpf/tuto_pipeline_v1.json",
    pipeline_root = "gs://vertex_chameleon_data"
).run()

ソース全体

ソースの全体は下記のような感じです。

from kfp import dsl #パイプライン作成用
from kfp.v2 import compiler #コンパイラ用
from kfp.v2.dsl import component #コンポーネント作成用
import google.cloud.aiplatform as aip


# コンポーネントの作成
# 手順1:処理を行うコンポーネント(関数)の作成
@component(base_image="python:3.7")
def say_message(message:str):
    print(f"{message}")


# パイプラインの作成
# 手順2:パイプラインの作成
@dsl.pipeline(
    name="tutorial-pipeline1"
)
def tutorial_pipeline():
    say_message(message="hello-world")
    pass


# コンパイル
# 手順3:コンパイルする対象のパイプラインと出力json名を指定してコンパイルを実行
compiler.Compiler().compile(
    pipeline_func = tutorial_pipeline,
    package_path  = "/home/npprw160/Sandbox/kpf/tuto_pipeline_v1.json",
)


# パイプラインの実行
# 手順4:コンパイルしたパイプラインの実行
aip.init(
    project="awesome-height-240603",
    location="us-central1"
)

aip.PipelineJob(
    display_name = "Pipeline基礎編",
    template_path = "/home/npprw160/Sandbox/kpf/tuto_pipeline_v1.json",
    pipeline_root = "gs://vertex_chameleon_data"
).run()