![コード
from fastapi import FastAPI
from src.app.routers import predict_api
app = FastAPI()
app.include_router(
predict_api.router,
pre?x='/predict',
tags=['predict'])
from fastapi import APIRouter
from src.app.api import _predict
from src.app.ml.classi?er import Data, Classi?er
router = APIRouter()
classi?er = Classi?er('model_path')
@router.get('/labels')
def labels():
return {'labels': Data().labels}
@router.get('')
def test():
return classi?er.predict(Data().test_data)
@router.post('')
def predict(data: Data):
return {'prediction': classi?er.predict(data)}
import numpy as np
import onnxruntime as rt
class Classi?er():
def __init__(self, model_path):
self.classi?er = None
self.load_model(model_path)
def load_model(self, model_path):
~~
def predict(self, input):
_prediction = self.classi?er.run(
None,
{self.input_name: input.astype('?oat32’)}
)
return _prediction
main.py
predict_api.py classi?er.py
GET /predictで
テストデータの
推論を得られる
ようにしておく。
入力の型を
明示する](https://image.slidesharecdn.com/mlsysteminpython-200828031911/85/Ml-system-in_python-34-320.jpg)
![推論器のインターフェイス
Docker image
Runtime
Model
入力
前処理
推論
出力
一度は経験する間違い：?oat32で学習した
モデルを?oat16で推論して結果がおかしい。
REST API + Json
● Webではポピュラーな実装。
● サーバ側でJsonを定義するが、型が
曖昧になるため、入力値の値確認(assertion)
が必要。
GRPC + Protocol Buffer
● TF ServingやONNX Runtime Serverで
標準提供。
（両者ともREST API + Jsonも標準提供）
● .protoファイルで入出力の型含めて
定義可能。クライアント、サーバで
共通化できる。
値の範囲や形式
● 正常に推論可能なデータの範囲。
● 自然言語の言語。
● 画像のフォーマット、RGB、その他。
{input_data: []}
((1,3,224,224))
?oat32
{prediction: []}
(1000,)
?oat32
Node pool
Node
LB](https://image.slidesharecdn.com/mlsysteminpython-200828031911/85/Ml-system-in_python-35-320.jpg)
![コード
from fastapi import APIRouter
from src.app.api import _predict
from src.app.ml.classi?er import Data,
Classi?er
router = APIRouter()
classi?er = Classi?er('model_path')
@router.get('/labels')
def labels():
return {'labels': Data().labels}
@router.post('')
async def predict(data: Data):
return {'prediction': classi?er.predict(data)}
import numpy as np
import grpc
import tensor?ow as tf
from tensor?ow_serving.apis import predict_pb2
from tensor?ow_serving.apis import
prediction_service_pb2_grpc
TFS_GPRC = 'prep_pred_tfs:8510'
class Classi?er():
def __init__(self, models):
self.models = models
self.prep = None
self.post = None
self.input = 'keras_layer_input'
self.input_shape = [1,224,224,3]
self.channel = 'model_spec_name'
self.stub = 'model_spec_name'
self.spec_name = 'model_spec_name'
self.signature_name = 'default'
class Classi?er():
def predict(self, input: Image):
input = np.array(self.prep.transform(input))
r = predict_pb2.PredictRequest()
r.model_spec.name = self.spec_name
r.spec.signature_name = self.signature_name
r.inputs[self.input].CopyFrom(
tf.make_tensor_proto(
input,
shape=self.input_shape))
result = self.stub.Predict(request, 5.0)
result = result.outputs[output_name].?oat_val
prediction = self.post.transform(result)
return prediction
routers.py
classi?er.py classi?er.py
前処理
→TF Serving (GRPC)
→後処理](https://image.slidesharecdn.com/mlsysteminpython-200828031911/85/Ml-system-in_python-39-320.jpg)
![コード
from fastapi import FastAPI
from src.app.routers import proxy
app = FastAPI()
app.include_router(
proxy.router,
pre?x='/redirect',
tags=['redirect'])
router = APIRouter()
class Data():
data: Any = None
async def _get_redirect(session, alias, url):
async with session.get(url) as r:
response_json = await r.json()
return {alias: {'response': response_json}}
@router.get('/{redirect_path:path}')
async def get_redirect(redirect_path):
async with aiohttp.ClientSession() as session:
tasks = [asyncio.ensure_future(
_get_redirect(
session, alias,
helpers.redirect_builder(
alias,url,redirect_path,redirect_map))
) for alias, url in urls.items()]
return await asyncio.gather(*tasks)
async def _post_redirect(
session, data, alias, url):
async with session.post(url, data) as r:
response_json = await r.json()
return {alias: {'response': response_json}}
@router.post('/{redirect_path:path}')
async def post_redirect(redirect_path, data):
data.data['job_id'] = get_job_id()
async with aiohttp.ClientSession() as session:
tasks = [asyncio.ensure_future(
_post_redirect(
session, data.data, alias,
helpers.redirect_builder(
alias,url,redirect_path,redirect_map))
) for alias, url in urls.items()]
return await asyncio.gather(*tasks)
main.py
proxy.py proxy.py
転送先は環境変数で
定義してdictに変換。
aiohttpでasyncリクエスト。](https://image.slidesharecdn.com/mlsysteminpython-200828031911/85/Ml-system-in_python-42-320.jpg)
![K8s Manifest
apiVersion: apps/v1
kind: Deployment
metadata:
name: api-proxy
labels:
app: api-proxy
namespace: serving-pattern
spec:
selector:
matchLabels:
app: api-proxy
replicas: 4
template:
metadata:
labels:
app: api-proxy
name: api-proxy
spec:
containers:
- name: api-proxy
image: api_proxy:latest
command: ["./run_proxy.sh"]
env:
- name: SVC_SKLEARN
value:
sklearn.serving.svc.cluster.local:8893
- name: SVC_RESNET_ONNX_SERVER
value:
resnet-server.serving.svc.cluster.local:8896
- name: REDIRECT_MAP
value: '{
"SVC_RESNET_ONNX_SERVER":{
"predict/label":"predict/async",
"predict":"predict/async"
}
}'
proxy.yaml proxy.yaml
Sklearn
WebSingleA
Resnet50
AsyncB
Proxy
キャッシュ
proxy.svc/predict
sklearn.svc/predict
onnx.svc/predict/async
転送先の定義例。
わかりやすく改行を
入れている。
（本当はcon?gmapで
定義）](https://image.slidesharecdn.com/mlsysteminpython-200828031911/85/Ml-system-in_python-43-320.jpg)
![コード
from fastapi import FastAPI
from src.app.routers import abtest
app = FastAPI()
app.include_router(
abtest.router,
pre?x='/abtest,
tags=[abtest])
@router.post('/{redirect_path:path}')
async def post_redirect(redirect_path, data):
data.data['job_id'] = get_job_id()
if data.ab in test_group.keys():
group = test_group[data.ab]
redirect = {group: redirect[group]}
async with aiohttp.ClientSession() as session:
tasks = [
asyncio.ensure_future(
_post_redirect(
session, data.data, alias,
helpers.redirect_builder(
alias, url, redirect_path, redirect))
) for alias, url in urls.items() if alias in
redirect.keys()]
responses = await asyncio.gather(*tasks)
return responses
apiVersion: apps/v1
kind: Deployment
~~
spec:
containers:
- name: abtest-proxy
image: abtest-proxy:latest
env:
- name: SERVICE_AB_TEST_A
value:
sklearn.serving.svc.cluster.local:8893
- name: SERVICE_AB_TEST_B
value:
resnet-server.serving.svc.cluster.local:8896
- name: ABTEST_GROUP
value:
'{"GROUP_A":"SERVICE_AB_TEST_A",
"GROUP_B":"SERVICE_AB_TEST_B",
"DEFAULT":"SERVICE_AB_TEST_A"}'
main.py
abtest.py proxy.yaml
ABテスト定義例。
わかりやすく
改行を入れている。
本来はcon?gmap](https://image.slidesharecdn.com/mlsysteminpython-200828031911/85/Ml-system-in_python-56-320.jpg)
More Related Content
What's hot (20)
Similar to Ml system in_python (20)
More from yusuke shibui (9)
Ml system in_python
- 2. 自己紹介 shibui yusuke ● 機械学習エンジニア & インフラエンジニア & ARエンジニア ● もともとクラウド基盤の開発、運用。 ● ここ4年くらいMLOpsで仕事。 ● Github: @shibuiwilliam ● Qiita: @cvusk ● FB: yusuke.shibui ● 最近やってること: Edge AIとAR cat : 0.55 dog: 0.45 human : 0.70 gorilla : 0.30 物体検知
- 3. 本日の内容 ● 机械学习のシステムを作るためのアーキテクチャや実装例を説明します。 ● 主に私がMLOpsやシステム開発、運用で成功、失敗してきたことをもとに話します。 ● ターゲットオーディエンス： ○ これから机械学习のシステム開発、運用を始めるソフトウェアエンジニア ○ 機械学習モデルを本番システムに組み込みたい機械学習エンジニア
- 4. MLシステムデザインパターン ● 以前から考えていた機械学習システムの作り方をパターン化して公開しました。 「デザインパターン」と呼ぶには偉大な先人に畏れ多い。。。 ● https://github.com/mercari/ml-system-design-pattern ● 上記はアーキテクチャ集になっていますが、実際に独自に作ってみたものが以下になります。 ● https://github.com/shibuiwilliam/ml-system-in-action ● 本セッションでは上記デザインパターンと実装例をもとに、主にPythonで機械学習システムを 作る方法を共有します。
- 5. Agenda 1. 机械学习のシステム 2. 机械学习システムをパターン化する 3. 実際に作ってみる 4. まとめ
- 6. 机械学习のシステム
- 7. 机械学习のシステム ● ユーザから見たAI ● MLエンジニアから見たML ● ソフトウェアエンジニアから見たMLシステム 猫！ 犬！ 猫！ https://papers.nips.cc/paper/5656-hidden-technical-debt-in-machine-learning-systems.pdf Accuracy 99.99%
- 8. ユーザに使ってもらえる機械学習へ https://pycon.market.com/ . . . Python 機械学習 cart acco unt help 商品A 商品B 商品D商品C 商品 カテゴリ 1 商品 カテゴリ 2 商品 カテゴリ 4 商品 カテゴリ 3 ブラウザ標準 パーソナライズ レコメンデーション 固定化された デザイン 検索欄の入力補完 自然言語処理 キャンペーン あるECサイト
- 12. 机械学习システムをパターン化する
- 18. モデルを作る
- 19. アンチパターン：Only meパターン ● コード、データ、モデルが管理されず、開発者の 環境でしか再現できない。 （開発者の環境でも再現できないこともある） ● DVCやML Metadataでステップごとにアウトプットをコミット することを推奨。 ● 推論のインターフェイス（入力データの型と shape、出力データの型とshape、意味）および サンプルデータを定義しておく。 import load local data evaluate retrain de?ne ml train save model git no training data 99.99% ???
- 20. 学習パイプラインとバージョニング データ取得 前処理 学習 評価 ビルド ● データパイプラインやMLパイプラインのワークフローは DAG（有向非巡回グラフ）として定義。 ● DWHやストレージから取得したデータを各フローで処 理し、結果をモデルレポジトリに記録。 ● モデル生成後にビルド。ビルドのアウトプットは実行環 境含めてDockerイメージとしてビルドすることもあれ ば、モデルのみをビルドし、実行環境はメタデータとし て定義することもある。 ● メタデータにはモデルの学習、評価に使った再現可能 なデータ、パラメータ、評価値を登録。 DWH ストレージ Hold out test メタデータ モデルレポジトリ Image builder Docker registry Docker registry パラメータ チューニング
- 21. 学習パイプラインとバージョニング DVC OSSのデータ?MLワークフロー管理。 Gitとともにローカルで実行する構成。 PythonSDKあり。 Google ML Metadata Kube?owやTFXのML Metadata管理。 メインはクラウド+Kubernetes。ローカルもサポート。 PythonSDKあり。 MetadataStore Polyaxon、ML?ow、Meta?ow、SageMakerも似通った構成を取っている。
- 22. バッチ学習 ● 学習パイプラインを定期的に実行する構成。 ● 学習データが時系列で変化し、定期的に モデルを更新する用途で有効。 ● 学習パイプラインをcronやジョブサーバから トリガーする構成。 ● 学習データは実行時に最新のものを使用する 一方、テストデータを更新するかは要検討。 Metadata cron: 0 0 1 * * DWH Hold out test Docker image Runtime Model データ取得 前処理 学習 評価 ビルド
- 23. モデルをリリースする
- 24. 学習環境と推論環境 Docker image Runtime Model 入力 前処理 推論 出力 データ取得 前処理 学習 評価 ビルド Docker image Library 学習環境のみ - Jupyter Notebook - バッチ学習用のライブラリ （例：PyTorch、TensorFlow、Keras） - バッチ学習用のインフラ（GPUとか） - データ?モデルバージョニングツール - モデルビルダー 両環境に共通 - モデルファイル - 前処理で使うライブラリ （例：Sklearn、PIL、Mecab） - 入出力のデータ型と形 推論環境のみ - 推論用のライブラリ （例：ONNX Runtime, TF Serving） - ランタイムと外部インターフェイス - ロガーとモニタリングツール
- 25. アンチパターン：Incompatible library and model ● 学習環境と推論器でライブラリのバージョンが違って しまい、推論器を起動できない状態。 例：.pklをロードできない状態。 または推論器を起動できるけど結果が間違っている状態。 例：ライブラリのバージョンによって演算結果が 違うことがある。 ● 推論器の起動時jobとしてテストデータの推論を実行し、想定通 りの結果になることを確認する。 ● 学習環境と推論環境のライブラリバージョンをあわせる 必要がある。 Docker image Runtime Model ビルド Base image Image builder 推論器 推論器 Runtime Image registry 推論 テスト sklearn 0.21.0 sklearn 0.23.0 sklearn 0.21.0
- 26. モデルの配布と推論器の稼働 ● 学習したモデルを推論器としてリリースし 稼働させるためにはモデルファイルだけで なく実行環境（OS、ランタイム）が必要。 ● モデルの配布方法は大きく2とおり。 a. モデル含めて推論器をビルド。 b. モデルと推論器を別にビルド。 ● ビルド方式の違いは推論器を起動させる 方法の違いとなる。 Docker image Runtime Model ビルド Base image Image builder Model repository 推論器 推論器 Runtime Image registry
- 27. モデルインイメージパターン ● 推論器のイメージにモデルを含めてビルド するパターン。 ● バージョン不一致によるトラブルを避ける ことが可能。 ● Dockerイメージサイズは大きくなるため、Docker pullに時間を要する。 ベースイメージを共通化することで pull時間を短縮することが可能。 ● Kubernetesやクラウドでスケーラブルな 推論システムを作る場合、Docker pull時間が オートスケールのボトルネックになる。 共通のベースイメージは全ノードにpullして 改善を図る。 ビルド Image builder Docker image Runtime Model Base image Runtime Image registry 推論器 Model repository
- 28. モデルロードパターン ● モデルファイルと推論器イメージを別に 管理するパターン。 Dockerイメージを共通かつ軽量に保つことが 可能。 ● 学習時と推論イメージでソフトウェアの バージョン不一致が発生する可能性が残る。 学習環境と推論イメージでバージョニング する必要がある。 ● 推論器起動時にDockerイメージをpull、run後、モデ ルファイルをダウンロードして稼働 させる。 KubernetesであればinitContainerを使用。 ビルド Model repository Base image Runtime 推論器
- 29. コード FROM python:3.8.5-slim ENV ML_DIR ml WORKDIR /${ML_DIR} ADD requirements.txt /${ML_DIR}/ RUN apt-get -y update && apt-get -y install apt-utils gcc && pip install --no-cache-dir -r requirements_api.txt COPY ./src/ /${ML_DIR}/src/ COPY ./logging/ /${ML_DIR}/logging/ COPY ./src/app/ml/model/* /${ML_DIR}/src/app/ml/model/ COPY ./src/app/ml/data/* /${ML_DIR}/src/app/ml/data/ spec: containers: - name: ml-api image: ml_api:latest volumeMounts: - name: model-volume mountPath: /model initContainers: - name: ml-download image: ml_download:latest env: - name: model_url value: https://aaa.pycon.jp/model.onnx volumeMounts: - name: model-volume mountPath: /model volumes: - name: model-volume Model in Docker?le Model load k8s manifest initContainerで モデルを ダウンロード モデル含めて Dockerイメージを ビルド
- 30. モデルの配布とスケールアウト Node pool Node モデルレポジトリ Image builder Docker registry Container Runtime Model Container Runtime Init container Model Model LB モデルインイメージ イメージサイズは増加するが、一度nodeに pull されていれば稼働は速い。 モデルとイメージのバージョニングを同時に 管理可能。 モデルロード コンテナ起動時にモデルをダウンロード。 モデルを頻繁に入れ替える場合に有効。 モデルとイメージのバージョンが分かれる。
- 31. システムを作る
- 32. WFH with Catパターン ● ネコとともに在宅勤務するパターン。 ● 仕事中、食事中、睡眠中、あらゆる状況で インタラプトされる。 ● 作業を中断してネコが満足するまで撫でる 必要がある。 ● 撫でずに無視すると以下のような障害を招く。 ○ キーボード歩き。 ○ 椅子爪研ぎ。 ○ ゴミ箱倒し。 ● 多頭飼いの場合、他ネコも平等に撫でないと 上記の障害を招く。 →安定稼働しているシステムでも不意にアラートが 発生する。 →ひとつ対策を立ててもまた別の課題が生じる。
- 33. Webシングルパターン ロードバランサー REST API Docker Python image Gunicorn Uvicorns FastAPI Model ● 最もシンプルな構成の推論器。 ● すぐリリースしたいときに有効。 ● GunicornやUWSGIをベースに FlaskやFastAPIで稼働。 Gunicorn(WSGI) Uvicorn (ASGI) Uvicorn (ASGI) Gunicorn(WSGI) Flask Flask
- 34. コード from fastapi import FastAPI from src.app.routers import predict_api app = FastAPI() app.include_router( predict_api.router, pre?x='/predict', tags=['predict']) from fastapi import APIRouter from src.app.api import _predict from src.app.ml.classi?er import Data, Classi?er router = APIRouter() classi?er = Classi?er('model_path') @router.get('/labels') def labels(): return {'labels': Data().labels} @router.get('') def test(): return classi?er.predict(Data().test_data) @router.post('') def predict(data: Data): return {'prediction': classi?er.predict(data)} import numpy as np import onnxruntime as rt class Classi?er(): def __init__(self, model_path): self.classi?er = None self.load_model(model_path) def load_model(self, model_path): ~~ def predict(self, input): _prediction = self.classi?er.run( None, {self.input_name: input.astype('?oat32’)} ) return _prediction main.py predict_api.py classi?er.py GET /predictで テストデータの 推論を得られる ようにしておく。 入力の型を 明示する
- 35. 推論器のインターフェイス Docker image Runtime Model 入力 前処理 推論 出力 一度は経験する間違い：?oat32で学習した モデルを?oat16で推論して結果がおかしい。 REST API + Json ● Webではポピュラーな実装。 ● サーバ側でJsonを定義するが、型が 曖昧になるため、入力値の値確認(assertion) が必要。 GRPC + Protocol Buffer ● TF ServingやONNX Runtime Serverで 標準提供。 （両者ともREST API + Jsonも標準提供） ● .protoファイルで入出力の型含めて 定義可能。クライアント、サーバで 共通化できる。 値の範囲や形式 ● 正常に推論可能なデータの範囲。 ● 自然言語の言語。 ● 画像のフォーマット、RGB、その他。 {input_data: []} ((1,3,224,224)) ?oat32 {prediction: []} (1000,) ?oat32 Node pool Node LB
- 36. 非同期推論パターン メッセージ 推論器 ● 推論結果を同期的に得る必要がない場合に 有効なパターン。 ● クライアントと推論器の中間にApache Kafkaや RabbitMQのようなメッセージングを配置する ことで実現。 ● 非同期の方式はリクエスト非同期型、 Publish/Subscribe型、Noti?cation型等がある。 ワークフローに応じて選択。 ● ただし重たいディープラーニングでは、 ユーザを待たせないために非同期とし、 許容遅延時間内のみポーリングすることも 可能。 Docker Python image multiprocessing Model Runtime Client クライアント
- 37. アンチパターン：All in oneパターン ロードバランサー REST API Docker Python image Sklearn ML Model Tensor?ow Image Model ONNX Runtime NLP Model ● 多様なモデルを同一コンテナに詰め込む。 ○ Docker image size: 10GB ○ CPU: 32コア ○ RAM: 64GB ● イメージサイズが増大し、パフォーマンスの ボトルネックや障害対応が複雑化する。 ● 前処理含めて複数のRuntimeを稼働させることにな り、ライブラリのバージョン管理も困難。 ● マルチモーダルでfusionする場合も様々な モデルの混合になるが、複数モードの入力と 特徴抽出を持つ1モデルに集約して同一Runtime で稼働させる。 TF Serving Image extraction Text extraction Fusion Predict
- 38. Prep Predパターン LB Preprocess ● 前処理と推論で必要なライブラリやリソースが 違うときに有効。 前処理はPython、推論はTF Serving。 ● 前処理と推論で異なる実行方式になり、 それぞれを異なるサービスとして稼働させる。 ● Tensor?owは学習と推論で実行方式が異なる。 ○ 学習ではtf.keras.?t()。 ○ 推論ではCPUでTF Serving。 ● ONNXもRuntime server(Beta)のほうが パフォーマンスは安定する。 ● ただし、学習時の前処理とモデルは一致する 必要があるため、前処理と推論を個別に更新する ことは避ける。 LB Docker Python image Gunicorn Uvicorn FastAPI Sklearn Client Docker TF serving image TF savedmodel Prediction
- 39. コード from fastapi import APIRouter from src.app.api import _predict from src.app.ml.classi?er import Data, Classi?er router = APIRouter() classi?er = Classi?er('model_path') @router.get('/labels') def labels(): return {'labels': Data().labels} @router.post('') async def predict(data: Data): return {'prediction': classi?er.predict(data)} import numpy as np import grpc import tensor?ow as tf from tensor?ow_serving.apis import predict_pb2 from tensor?ow_serving.apis import prediction_service_pb2_grpc TFS_GPRC = 'prep_pred_tfs:8510' class Classi?er(): def __init__(self, models): self.models = models self.prep = None self.post = None self.input = 'keras_layer_input' self.input_shape = [1,224,224,3] self.channel = 'model_spec_name' self.stub = 'model_spec_name' self.spec_name = 'model_spec_name' self.signature_name = 'default' class Classi?er(): def predict(self, input: Image): input = np.array(self.prep.transform(input)) r = predict_pb2.PredictRequest() r.model_spec.name = self.spec_name r.spec.signature_name = self.signature_name r.inputs[self.input].CopyFrom( tf.make_tensor_proto( input, shape=self.input_shape)) result = self.stub.Predict(request, 5.0) result = result.outputs[output_name].?oat_val prediction = self.post.transform(result) return prediction routers.py classi?er.py classi?er.py 前処理 →TF Serving (GRPC) →後処理
- 40. 並列マイクロサービスパターン Client ● 独立した複数の推論モデルに並行して リクエストするときに有効。 例：複数の2値分類モデルで推論したいとき。 例：マルチモーダルで推論器を各モードに 分けるとき。 ● クライアントと推論器の中間にプロキシを 置き、プロキシで各推論器へのリクエストを 制御する構成。 ● リクエスト先を環境変数等で制御すると、 推論器の追加、削除を柔軟に実現することが 可能。 ● Nginxを仲介にしても良いが、推論結果の 集計やクライアントへのレスポンスに ロジックを持たせる場合はProxyを自作する。Docker Python image Gunicorn aiohttp server aiohttp client LB Web single A Prep pred C Web single B Proxy
- 41. マルチステージ推論パターン Client ● 複数の推論器でレイテンシーに差がある場合、 速い推論器は同期的に返し、遅い推論器の レスポンスを非同期にする。 ● クライアントと推論器の中間にProxyを置き、Proxyで各 推論器へのリクエストを制御する構成。 ● 推論器は同期、非同期両方のエンドポイントを 公開しておくことで、より柔軟な対応が可能。 この構成の場合、メッセージングではなく CeleryやBackgroundTasksで実装。 LB Sklearn WebSingleA Deep learning AsyncB Proxy キャッシュ Environment Variable - SVC_A: WebSingleA - SVC_B: AsyncB - SVC_A_EP: predict - SVC_B_EP: async
- 42. コード from fastapi import FastAPI from src.app.routers import proxy app = FastAPI() app.include_router( proxy.router, pre?x='/redirect', tags=['redirect']) router = APIRouter() class Data(): data: Any = None async def _get_redirect(session, alias, url): async with session.get(url) as r: response_json = await r.json() return {alias: {'response': response_json}} @router.get('/{redirect_path:path}') async def get_redirect(redirect_path): async with aiohttp.ClientSession() as session: tasks = [asyncio.ensure_future( _get_redirect( session, alias, helpers.redirect_builder( alias,url,redirect_path,redirect_map)) ) for alias, url in urls.items()] return await asyncio.gather(*tasks) async def _post_redirect( session, data, alias, url): async with session.post(url, data) as r: response_json = await r.json() return {alias: {'response': response_json}} @router.post('/{redirect_path:path}') async def post_redirect(redirect_path, data): data.data['job_id'] = get_job_id() async with aiohttp.ClientSession() as session: tasks = [asyncio.ensure_future( _post_redirect( session, data.data, alias, helpers.redirect_builder( alias,url,redirect_path,redirect_map)) ) for alias, url in urls.items()] return await asyncio.gather(*tasks) main.py proxy.py proxy.py 転送先は環境変数で 定義してdictに変換。 aiohttpでasyncリクエスト。
- 43. K8s Manifest apiVersion: apps/v1 kind: Deployment metadata: name: api-proxy labels: app: api-proxy namespace: serving-pattern spec: selector: matchLabels: app: api-proxy replicas: 4 template: metadata: labels: app: api-proxy name: api-proxy spec: containers: - name: api-proxy image: api_proxy:latest command: ["./run_proxy.sh"] env: - name: SVC_SKLEARN value: sklearn.serving.svc.cluster.local:8893 - name: SVC_RESNET_ONNX_SERVER value: resnet-server.serving.svc.cluster.local:8896 - name: REDIRECT_MAP value: '{ "SVC_RESNET_ONNX_SERVER":{ "predict/label":"predict/async", "predict":"predict/async" } }' proxy.yaml proxy.yaml Sklearn WebSingleA Resnet50 AsyncB Proxy キャッシュ proxy.svc/predict sklearn.svc/predict onnx.svc/predict/async 転送先の定義例。 わかりやすく改行を 入れている。 （本当はcon?gmapで 定義）
- 44. パラメータベース推論パターン Client ● 独立した複数の推論モデルに並行して リクエストするときに有効。 ● リクエスト先をProxyの環境変数等で 制御する。リクエストの挙動も定義可能。 ○ 例：接続先と割合の定義 ○ 例：タイムアウトやリトライ数 ● 個々の推論器に固有のパラメータを 設定することも可能だが、複雑になる 場合は別の推論システムにする。 LB WebSingleA PrepPredC Environment Variable - SVC_A: WebSingleA - SVC_B: WebSingleB - SVC_C: PrepPredC - SVC_A_RATE: 0.4 - SVC_B_RATE: 0.4 - SVC_C_RATE: 0.2 - TIMEOUT_SEC: 1 - RETRY: 2 WebSingleB Proxy Environment Variable - THRESHOLD: 0.99 - WORKERS: 4
- 45. 推論サーキットブレーカーパターン Client ● 急激な負荷増でスケールアウトが 間に合わない場合、一部のリクエストを 遮断することで全断を防ぐ。 ● 機械学習、特にディープラーニング推論の 場合、使用するリソースや、モデルを 含むイメージサイズが大きい傾向にあり、 ノードを含めたスケールアウトに時間を 要することがある。 ● NginxやEnvoy proxyで標準的に提供されて いる機能。 ● パフォーマンス劣化対策はサーキット ブレーカー以外でも可能。 Gunicorn（wsgi）ではworker数やtimeout, max_requests等でリソース利用を効率化する 機構が備わっている。 LB Nginx limit_req_zone $server_addr zone=moderateReqs:1m rate=100r/s; LB
- 46. アンチパターン：Nobody knowsパターン LB ML ML ML ML ML ML ML? ● 稼働中の推論器がなぜ動いているのか誰も知らないパ ターン。歴史的にシステムが複雑になっていくと頻繁に発 生する。 ● MLプロジェクトの初期にプロトタイプで作った モデルや、緊急対応で稼働させた推論器を残して おくと「Nobody knows」になる。 ● 学習データが残っていない状態は珍しくない。 学習コードもJupyter Notebookのみという場合、 モデルの評価や再現が困難。ML? ML
- 48. Cat injectionパターン ● 作業途中でネコに椅子を奪われるパターン。 ● 30分以上ネコを放置していると、離席中に発生。 ● ネコの機嫌を評価する指標になる。
- 49. 機械学習システムの品質 機械学習システムの品質には3カテゴリある。 1. 機械学習の推論モデル a. 推論モデルのパフォーマンス b. 本番データの変化による劣化とエラー c. 問題定義とソリューション 2. 推論モデルを稼働させるシステム a. 入出力のデータやデータ型 b. 推論スピードと可用性 c. 例外処理 3. 運用と体制 a. モデルの再現および再学習 b. 推論器の再現およびロールバック c. 維持可能な運用体制 Client LB LB int ?oat test data accuracy: 99.99% 何のML だっけ？ 1sec/req モデル作った VM消したよ dockerimg:latest 上書き error率 0.01% アサイン 変わった me too 転 職 商品カテゴリ 追加削除 そして誰も いなくなった
- 50. 負荷テスト Load tester LB Load tester ● 推論器を本番同様のリソースで稼働させて レイテンシーを測り、ボトルネックを発見、 チューニングする。 ● Too bigなモデルでない限り、モデルは変更せずに 実行環境で高速化や可用性強化を図る。 ● 推論器のボトルネックは推論モデル。 Webサーバに組み込んで計測する前に、 モデル単体をサーバ内部で連続して 推論させ、推論の平均所要時間を計測する。 ● 学習コストの高いディープラーニングモデルを 使う場合、学習前にモデルだけで負荷テストを 実行して遅延を確認しておく。 Pro?le 入力 前処理 推論 後処理 出力 レイテンシー（ms） 計測時間 計測時間 リクエスト数／秒
- 51. 負荷テスト ● 負荷テスト ○ 外部から連続してリクエストを送信。 入力がサイズ不特定の画像やテキストの場合、 ランダムなサイズのデータを用意。 ○ CPU使用率が上がらず遅延が発生する場合、 ワーカー数や通信方法を調整する。 NumpyやSklearnであればIntel MKLを使用する ことも選択肢。 ○ メモリリークが発生して根本解決が困難な場合、 Gunicornのmax_requestsパラメータで、一定数の リクエスト後にワーカープロセスを再起動する ことで緩和可能。 jitterやgraceful-timeoutと組み合わせる。 ○ Tensor?owやPytorchであれば前処理含めて 計算グラフに組み込むことで遅延短縮可能。 ○ 推論用GPUはコスト対効果次第。 Load tester LB Load tester 計測時間 リクエスト数／秒 計測時間 リソース RAM CPU
- 53. Gunicorn + Flask Gunicorn + FastAPI Prep/Pred 負荷テスト Gunicorn Flask ONNX Runtime Gunicorn Uvicorns FastAPI ONNX Runtime Gunicorn Uvicorns FastAPI ONNX Client ONNX Runtime Server ONNX Runtime 同一リソース（CPU/RAM）。 20rpsで検証。 PytorchベースのResnet50 onnx。 1. Gunicorn + Flask(WebSingle) 2. Gunicorn + FastAPI(WebSingle) 3. Gunicorn+FastAPI& ONNX Runtime Server(Prep/pred)
- 54. シャドウABテスト Client LB WebSingleA WebSingleBProxy Environment Variable - SVC_A: WebSingleA - SVC_B: WebSingleB - ACTIVE: WebSingleA - SHADOW: WebSingleB ● 稼働中モデルの更新では以下の確認が必要。 a. 推論サーバの正常稼働 b. 想定通りの推論結果 c. レイテンシー d. 負荷耐性 ● 更新推論モデルをシャドウで本番システムに 組み込み、リクエストを送る （ただしレスポンスは返さない）。 ● 推論結果だけでなくレイテンシーもログに残す。稼働中 モデルとシャドウモデルのログを照合する 必要があるため、リクエストに一意のIDを振る。 {“test_id”: “001”, “request_id”: “aaa”, “user_id”: “111”, “data_id”: “abc”, “request”: “WebSingleA”, “result”: {“pred”: 0, “latency”: 10ms}} {“test_id”: “001”, “request_id “aaa”, “user_id”: “111”, “data_id”: “abc”, “request”: “WebSingleB”, “result”: {“pred”: 1, “latency”: 5ms}}
- 55. オンラインABテスト Client LB WebSingleA WebSingleBProxy Environment Variable - GROUP_A: SVC_A - GROUP_B: SVC_B - DEFAULT: SVC_A - UNTIL: 20200829 ● オンラインABテストではモデルの更新が エンドユーザに与える影響を測る。評価指標に 応じてログを取得。 ● Proxyでリクエスト先をコントロールする。 特定のユーザグループを比較する場合、 ユーザグループの分割とリクエスト先の ルールをProxyでコントロールする。 ● ABテスト期間を設定して無効化する。 期間後に不要な推論器を削除しないと 「Nobody knows」になる。 Group_A Group_B {“test_id”: “001”, “request_id”: “aaa”, “user_id”: “111”, “data_id”: “abc”, “request”: “WebSingleA”, “result”: {“pred”: 0, “latency”: 10ms}} {“test_id”: “001”, “request_id “zzz”, “user_id”: “000”, “data_id”: “def”, “request”: “WebSingleB”, “result”: {“pred”: 1, “latency”: 5ms}}
- 56. コード from fastapi import FastAPI from src.app.routers import abtest app = FastAPI() app.include_router( abtest.router, pre?x='/abtest, tags=[abtest]) @router.post('/{redirect_path:path}') async def post_redirect(redirect_path, data): data.data['job_id'] = get_job_id() if data.ab in test_group.keys(): group = test_group[data.ab] redirect = {group: redirect[group]} async with aiohttp.ClientSession() as session: tasks = [ asyncio.ensure_future( _post_redirect( session, data.data, alias, helpers.redirect_builder( alias, url, redirect_path, redirect)) ) for alias, url in urls.items() if alias in redirect.keys()] responses = await asyncio.gather(*tasks) return responses apiVersion: apps/v1 kind: Deployment ~~ spec: containers: - name: abtest-proxy image: abtest-proxy:latest env: - name: SERVICE_AB_TEST_A value: sklearn.serving.svc.cluster.local:8893 - name: SERVICE_AB_TEST_B value: resnet-server.serving.svc.cluster.local:8896 - name: ABTEST_GROUP value: '{"GROUP_A":"SERVICE_AB_TEST_A", "GROUP_B":"SERVICE_AB_TEST_B", "DEFAULT":"SERVICE_AB_TEST_A"}' main.py abtest.py proxy.yaml ABテスト定義例。 わかりやすく 改行を入れている。 本来はcon?gmap
- 57. まとめ
- 58. まとめ ● End to Endで机械学习のシステムを作る実装例を説明しました。 ● 機械学習を本番導入する注意点はいろいろありますが、有効活用すれば絶大な効果を 発揮できると思います。 ● 「モデルを作る」から「本番稼働させる」までの架け橋となると幸いです。