機械学習モデルの開発は、Jupyter Notebook上での実験からプロダクション環境へのデプロイまで、多くの工程を含みます。しかし、実験管理が属人的になったり、モデルのバージョン管理が曖昧になったりすると、再現性の確保やチームでの協業が困難になります。こうした課題を解決するのがMLOps（Machine Learning Operations）であり、その中核を担うのがMLOpsパイプラインです。
筆者はこれまで複数のMLプロジェクトでMLflowとKubeflowを使い分けてきました。本記事では、両ツールの特徴を比較しつつ、実際の構築手順を解説します。
MLflowは、Databricks社が開発したオープンソースのMLライフサイクル管理プラットフォームです。主に以下の4つのコンポーネントで構成されています。
導入の敷居が低いのがMLflowの最大の強みです。既存のPythonコードに数行追加するだけで実験管理を始められます。
入会申込
ログインimport mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score
import pandas as pd
# トラッキングサーバーの設定
mlflow.set_tracking_uri("http://mlflow-server:5000")
mlflow.set_experiment("customer-churn-prediction")
# データ準備
df = pd.read_csv("customer_data.csv")
X = df.drop("churned", axis=1)
y = df["churned"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 実験実行
with mlflow.start_run(run_name="rf_baseline"):
params = {
"n_estimators": 100,
"max_depth": 10,
"min_samples_split": 5,
}
mlflow.log_params(params)
model = RandomForestClassifier(**params, random_state=42)
model.fit(X_train, y_train)
predictions = model.predict(X_test)
accuracy = accuracy_score(y_test, predictions)
f1 = f1_score(y_test, predictions)
mlflow.log_metrics({"accuracy": accuracy, "f1_score": f1})
mlflow.sklearn.log_model(model, "model")
print(f"Accuracy: {accuracy:.4f}, F1: {f1:.4f}")MLflow Model Registryを使うと、モデルのライフサイクルを「Staging」「Production」「Archived」のステージで管理できます。これにより、どのモデルが本番稼働中で、どのモデルが検証中なのかが一目で分かります。
from mlflow.tracking import MlflowClient
client = MlflowClient()
# モデルの登録
result = mlflow.register_model(
model_uri=f"runs:/{run_id}/model",
name="churn-prediction-model"
)
# ステージの遷移
client.transition_model_version_stage(
name="churn-prediction-model",
version=result.version,
stage="Production"
)KubeflowはKubernetes上で動作するMLプラットフォームです。Kubeflow Pipelinesを中心に、分散学習、ハイパーパラメータチューニング、モデルサービングなど、大規模なML基盤に必要な機能を包括的に提供します。
MLflowと比較すると、導入コストは高いものの、スケーラビリティとワークフローの自動化においては優位性があります。Kubernetesクラスタが前提となるため、インフラチームとの連携が不可欠です。
from kfp import dsl
from kfp.dsl import component, Input, Output, Dataset, Model, Metrics
@component(base_image="python:3.11-slim", packages_to_install=["pandas", "scikit-learn"])
def preprocess_data(
input_path: str,
output_dataset: Output[Dataset],
):
import pandas as pd
df = pd.read_csv(input_path)
df = df.dropna()
df.to_csv(output_dataset.path, index=False)
@component(base_image="python:3.11-slim", packages_to_install=["pandas", "scikit-learn"])
def train_model(
dataset: Input[Dataset],
model_output: Output[Model],
metrics_output: Output[Metrics],
n_estimators: int = 100,
):
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import pickle
df = pd.read_csv(dataset.path)
X = df.drop("target", axis=1)
y = df["target"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
clf = RandomForestClassifier(n_estimators=n_estimators)
clf.fit(X_train, y_train)
accuracy = clf.score(X_test, y_test)
metrics_output.log_metric("accuracy", accuracy)
with open(model_output.path, "wb") as f:
pickle.dump(clf, f)
@dsl.pipeline(name="ml-training-pipeline")
def ml_pipeline(input_path: str = "gs://my-bucket/data.csv"):
preprocess_task = preprocess_data(input_path=input_path)
train_task = train_model(
dataset=preprocess_task.outputs["output_dataset"],
n_estimators=200,
)両ツールは競合するものではなく、併用も可能です。以下の基準で選定するとよいでしょう。
MLOpsパイプラインの構築は、機械学習プロジェクトの成熟度を高める重要な投資です。MLflowは導入のしやすさと柔軟性で、Kubeflowはスケーラビリティと包括的な機能で、それぞれ異なる強みを持っています。プロジェクトの規模やチームのスキルセットに応じて適切に選択し、段階的にMLOps基盤を整備していくことが成功への近道です。
