随時アップデートしているため中途半端なドキュメントです！その前提で読んでください。
概要
こんにちは、最近kubeflow気になっているマンです。
kubeflow のツールの1つに Kubeflow Fairing（フェアリング）というものがありますが、
日本語の情報があまりにも少なく、英語の公式ドキュメントも少ない上に散在しているという状況です。
個人的にはこのFairingはめちゃくちゃ便利であると睨んでいて、いつの日かプロダクション環境で使ってみたいという気持ちがあります。
なのでアップデートがあり次第追記していくというスタイルで kubeflow Fairing についての知見を貯めていこうと思います。結果として他の方が見てくださったり、または他の方が知った知見がここに集まったりすればいいなという期待を込めてお送りします。
Kubeflow Fairing とは
公式ドキュメント
kubeflow Fairing とは一体何でしょう？ 公式ドキュメント には以下のようにあります。
Kubeflow Fairing is a Python package that makes it easy to train and deploy ML models on Kubeflow. Kubeflow Fairing can also been extended to train or deploy on other platforms. Currently, Kubeflow Fairing has been extended to train on Google AI Platform.
MLなんでもてんこ盛りセットであるkubeflowの中でも、kubeflowやGCPへのMLモデルの学習やデプロイ（サービング）を手助けするPythonパッケージツールであると記述されています。
また以下のように続きます。
The following are the goals of the Kubeflow Fairing project:
• Easily package ML training jobs: Enable ML practitioners to easily package their ML model training code, and their code’s dependencies, as a Docker image.
• Easily train ML models in a hybrid cloud environment: Provide a high-level API for training ML models to make it easy to run training jobs in the cloud, without needing to understand the underlying infrastructure.
• Streamline the process of deploying a trained model: Make it easy for ML practitioners to deploy trained ML models to a hybrid cloud environment.
Kubeflow Fairing の目的を改めて日本語でざっくりとまとめると3つになります。
- MLにおける学習のジョブを簡単にDockerイメージなどにパッケージできること
- MLにおける学習を、あらゆる環境で 楽に 実行できること
- MLにおける学習→デプロイのハードルをもっと下げること
上記で書かれていることは新しいことではなく、MLをプロダクションに載せて運用していこうとすると誰しもがぶつかる問題だと思います。kubeflow では上記を達成するために近接するサービスがあります。
が、上記の課題を正面から取り扱うFairingがあればより楽にアーキテクチャ設計ができるのではないでしょうか。
Cloud Next '19
動画からのベタ貼りになりますが、画像の出典はすべて ML Ops Best Practices on Google Cloud (Cloud Next '19) となります。
ここからはスピーカーによるKubeflow Fairingの紹介と、私の解釈を書いていきます。
前提として複数の環境を組み合わせてMLサービスを開発、提供することを考えます。所謂MLOpsというやつです。
ユースケースとして以下のような環境を挙げます。ユーザはオンプレミスのKubeflow環境で、Jupyter Notebooksを使ってモデルを構築、学習、推論のモジュールを作成します。
そして開発環境で安定したパフォーマンスを出せることが分かったらそれをGCP（やAWS）の環境にて再現するでしょう。
AI Platform はGCP上で動くML何でもマネージドセットです。学習であればAI Platform Trainingを利用できますし、予測であればAI Platform Predictionを利用できます。
両環境の差異はあるものの、Python と Jupyter を利用していることは共通しています。
そのため両環境を行き来してそれぞれに特化したコードを書かせないために、Hybrid ML SDK が求められます。
Kubeflow Fairing は「データサイエンティストが」「どの環境にも共通して適用できるようなコードが書けること」をその目的としています。そしてそれこそが上記で言っていた Hybrid ML SDK です。
例えば今日、複数環境を跨いでモデルを構築、学習、サービングするとなると以下の図のようなコードを書くことが求められます。
- ローカル環境にてモデルをEDA的に構築する。ローカルPCのスペックは相対的に乏しいので、データを少なくする等の工夫も必要。
- GCPなどのマネージドサービスを利用して、学習や推論APIの作成をする。
- kubernetes 上にサービングするためにジョブのマニフェストファイルを書き、applyする。
kubernetesはMLモデルを構築したり、サービングするのにとても便利です。スケジューリングの機能もありますし、GPUのノードプールもうまく設定してやれば、より効率的にGPUの利用ができるようになります。
またGCPの各マネージドサービスを利用すれば、ノードプールやノード自体も管理する必要がなく、文字通りインスタンスにジョブを丸投げして、実行完了したらインスタンス削除するところまでをいい感じにやってくれます。
ただしAI Platform Trainingを利用した方は御存知の通り、便利な半面、ディレクトリ構成をGCPのドキュメント通りにしたり、リージョンの制限と戦ったりと巨人の方に乗るためにアジャストしないといけないことも多々あります。その結果以下のように、達成したいことは「MLモデルの構築→学習→サービング」なのに、それぞれにオーバーフィットしていまいがちです。
また上記のようなシンプルなモデルならこれで済みますが、各社がゴリッと書いたコードを適用するとなると…（白目）という感じでしょう。
長々と書きましたが、Kubeflow Fairingを利用するととそれらの差分を吸収してくれます。これこそがFairingです。
下の図のように引数として環境を渡してやれば、他は全く同じコードとして書くことができます。
そのため「DSとデータエンジニアの間の壁を超えて、DSのこれまでの作業の延長上でAPIサービングまでもっていける」というKubeflowの目的の達成にぐっと近づくことできます。
動画の中ではデモを行っています。
図は上記を再度まとめたものです。GCPの提供するサービスをうまく組み合わせることができれば、環境の差異を吸収しつつ、コスパよくスピード感をもってML機能をプロダクションに乗せることが可能になるはずです。
チュートリアル
kubeflowに項目として載っているFairingのドキュメントには残念ながら、豊富に情報がある状態とは言えません。
探してみるとGitHubにもexampleがありました。またSDKのドキュメントはコチラにありました。
exampleに沿ってGCP環境に推論APIをサービングすることをしてみましたが、メソッドの挙動がガラッと変わっており頓挫しております。(2020/05/17時点)
例えば、このexampleを参考にした時、
GCPServingDeployer() で引数を渡す必要があったり、そのインスタンスでメソッドを呼び出す時にコケたり、というようにexample通りに進めるがうまく行きません。
各モジュールやツールのバージョンの組み合わせが悪いような気もしますが、一旦は発展途上ということにしてペンドにしています。
# Define variables
PROJECT_ID = '<project-id>'
VERSION_DIR = 'gs://<bucket-name>/<folder-name>/'
MODEL_NAME = '<model-name>'
VERSION_NAME = '<version-name>'
# Define training and evaluation functions
import argparse
import pandas as pd
from sklearn.metrics import mean_absolute_error
from sklearn.model_selection import train_test_split
from xgboost import XGBRegressor
import urllib.request
try:
from sklearn.preprocessing import Imputer
except ImportError:
from sklearn.impute import SimpleImputer as Imputer
import logging
logging.getLogger().setLevel(logging.INFO)
TRAINING_URL="https://raw.githubusercontent.com/kubeflow/examples/master/xgboost_ames_housing/ames_dataset/train.csv"
TRAINING_FILE="train.csv"
ESTIMATORS=1000
LEARNING_RATE=0.1
TEST_FRACTION_SIZE=0.25
EARLY_STOPPING_ROUNDS=50
def run_training_and_eval():
(train_X, train_y), (test_X, test_y) = read_input()
model = train_model(train_X,
train_y,
test_X,
test_y,
ESTIMATORS,
LEARNING_RATE)
eval_model(model, test_X, test_y)
def download(url, file_name):
with urllib.request.urlopen(url) as response, open(file_name, "wb") as file:
file.write(response.read())
def read_input(test_size=TEST_FRACTION_SIZE):
"""Read input data and split it into train and test."""
download(TRAINING_URL, TRAINING_FILE)
data = pd.read_csv(TRAINING_FILE)
data.dropna(axis=0, subset=['SalePrice'], inplace=True)
y = data.SalePrice
X = data.drop(['SalePrice'], axis=1).select_dtypes(exclude=['object'])
train_X, test_X, train_y, test_y = train_test_split(X.values,
y.values,
test_size=test_size,
shuffle=False)
imputer = Imputer()
train_X = imputer.fit_transform(train_X)
test_X = imputer.transform(test_X)
return (train_X, train_y), (test_X, test_y)
def train_model(train_X,
train_y,
test_X,
test_y,
n_estimators,
learning_rate):
"""Train the model using XGBRegressor."""
model = XGBRegressor(n_estimators=n_estimators,
learning_rate=learning_rate)
model.fit(train_X,
train_y,
early_stopping_rounds=EARLY_STOPPING_ROUNDS,
eval_set=[(test_X, test_y)])
logging.info("Best RMSE on eval: %.2f with %d rounds",
model.best_score,
model.best_iteration+1)
return model
def eval_model(model, test_X, test_y):
"""Evaluate the model performance."""
predictions = model.predict(test_X)
logging.info("mean_absolute_error=%.2f", mean_absolute_error(predictions, test_y))
# Train and evaluate the model locally
(train_X, train_y), (test_X, test_y) = read_input()
model = train_model(train_X,
train_y,
test_X,
test_y,
ESTIMATORS,
LEARNING_RATE)
eval_model(model, test_X, test_y)
# Export the model using the joblib library
import joblib
joblib.dump(model, 'model.joblib')
!gsutil cp model.joblib {VERSION_DIR}
# Deploy the model to GCP
from kubeflow.fairing.deployers.gcp.gcpserving import GCPServingDeployer
deployer = GCPServingDeployer()
deployer.deploy(VERSION_DIR, MODEL_NAME, VERSION_NAME)
# Send a prediction to the deployed model
from googleapiclient import discovery
ml = discovery.build('ml', 'v1')
resource_name = 'projects/{}/models/{}/versions/{}'.format(PROJECT_ID, MODEL_NAME, VERSION_NAME)
ml.projects().predict(
name=resource_name,
body={
'instances': [
[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37]
]
}
).execute()
最後に
時間が経てば情報量も増えてきたり、バージョンが安定したりしてくるはずなので、随時記事をアップデートしていきます。何か情報があれば教えて下さいmm