LabHub

ブログ

Kubeflow Pipelines MLワークフローオーケストレーション実践ガイド：KFP v2 SDKからプロダクションデプロイまで

2026-03-11 · 16 分で読めます · #ai-platform#kubeflow#mlops#pipeline-orchestration#kubernetes#2026-03

한국어English日本語中文

Kubeflow Pipelines ML ワークフローオーケストレーションガイド

はじめに

ML プロジェクトがプロダクション規模へ成長すると、データ前処理からモデル学習、評価、デプロイまでのワークフロー全体を安定して管理することが中心的な課題になる。Jupyter Notebook での実験は再現性が低く、手動でのスクリプト実行はミスが起きやすい。こうした問題を解決するために ML パイプラインオーケストレーション ツールが必要になる。

Kubeflow Pipelines(KFP) は Google が主導するオープンソースプロジェクトで、Kubernetes 上で ML ワークフローを定義し実行するプラットフォームである。各ステップを独立したコンテナとして実行することで再現性を保証し、パイプラインのバージョン管理と実験の追跡に対応する。本記事では KFP v2 SDK のアーキテクチャから実践的なパイプライン構築、プロダクション運用戦略までを詳しく扱う。

Kubeflow Pipelines のアーキテクチャ

中心的なコンポーネント構造

Kubeflow Pipelines は複数のマイクロサービスで構成される。

コンポーネント役割技術スタック
Pipeline Serviceパイプラインの CRUD、実行管理gRPC/REST API
Metadata Serviceアーティファクトと実行メタデータの保存ML Metadata (MLMD)
Persistence Agentワークフロー状態を DB へ同期Kubernetes Controller
Scheduler繰り返し実行(Recurring Run)の管理CronJob ベース
UI ServerWeb ダッシュボードReact ベース SPA
Artifact Storeパイプライン成果物の保存MinIO / S3 / GCS

KFP v2 でのアーキテクチャ変更点

KFP v2 では v1 に比べて根本的なアーキテクチャ変更が行われた。従来の Argo Workflows への依存を取り除き、独自のワークフローエンジンを導入している。

# KFP v2 vs v1 の主な違いの比較
"""
KFP v1:
- Argo Workflows ベースの実行
- kfp.dsl.ContainerOp を使用
- YAML ベースのパイプライン定義が可能

KFP v2:
- 独自のワークフローエンジン (または Argo を選択可能)
- kfp.dsl.component デコレータを使用
- IR (Intermediate Representation) YAML の導入
- ML Metadata のネイティブ統合
- 型安全なコンポーネントインターフェース
"""

# v2 のアーキテクチャレイヤー
ARCHITECTURE_LAYERS = {
    "SDK Layer": "Python DSL でパイプラインを定義 (kfp.dsl)",
    "IR Layer": "プラットフォーム非依存の中間表現 (PipelineSpec YAML)",
    "Backend Layer": "パイプラインの実行と管理 (API Server)",
    "Runtime Layer": "コンテナオーケストレーション (K8s Pod)",
    "Metadata Layer": "実行履歴とアーティファクトの追跡 (MLMD)",
}

KFP v2 SDK の基本的な使い方

コンポーネントの作成

KFP v2 では、コンポーネントを @component デコレータで定義する。各コンポーネントは独立したコンテナで実行される。

from kfp import dsl
from kfp.dsl import Input, Output, Dataset, Model, Metrics

# 軽量な Python コンポーネント (依存が少ない場合)
@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.1.4", "scikit-learn==1.4.0"],
)
def preprocess_data(
    raw_data_path: str,
    test_size: float,
    train_dataset: Output[Dataset],
    test_dataset: Output[Dataset],
    metrics: Output[Metrics],
):
    """データ前処理コンポーネント"""
    import pandas as pd
    from sklearn.model_selection import train_test_split

    df = pd.read_csv(raw_data_path)

    # 欠損値の処理
    df = df.dropna(subset=["target"])
    df = df.fillna(df.median(numeric_only=True))

    # 学習/テストの分割
    train_df, test_df = train_test_split(
        df, test_size=test_size, random_state=42, stratify=df["target"]
    )

    # アーティファクトとして保存
    train_df.to_csv(train_dataset.path, index=False)
    test_df.to_csv(test_dataset.path, index=False)

    # メトリクスのロギング
    metrics.log_metric("total_samples", len(df))
    metrics.log_metric("train_samples", len(train_df))
    metrics.log_metric("test_samples", len(test_df))
    metrics.log_metric("feature_count", len(df.columns) - 1)

カスタムコンテナコンポーネント

重い依存が必要な場合はカスタムコンテナイメージを使う。

# カスタムイメージベースのコンポーネント
@dsl.component(
    base_image="gcr.io/my-project/ml-training:v2.1",
)
def train_model(
    train_dataset: Input[Dataset],
    model_type: str,
    hyperparameters: dict,
    trained_model: Output[Model],
    metrics: Output[Metrics],
):
    """モデル学習コンポーネント"""
    import pandas as pd
    import joblib
    from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
    from sklearn.metrics import accuracy_score, f1_score

    train_df = pd.read_csv(train_dataset.path)
    X_train = train_df.drop("target", axis=1)
    y_train = train_df["target"]

    # モデルの選択
    model_map = {
        "random_forest": RandomForestClassifier,
        "gradient_boosting": GradientBoostingClassifier,
    }
    model_cls = model_map[model_type]
    model = model_cls(**hyperparameters)
    model.fit(X_train, y_train)

    # モデルの保存
    joblib.dump(model, trained_model.path)

    # 学習メトリクス
    y_pred = model.predict(X_train)
    metrics.log_metric("train_accuracy", accuracy_score(y_train, y_pred))
    metrics.log_metric("train_f1", f1_score(y_train, y_pred, average="weighted"))
    metrics.log_metric("model_type", model_type)

    # モデルのメタデータ
    trained_model.metadata["framework"] = "scikit-learn"
    trained_model.metadata["model_type"] = model_type

パイプラインの定義

コンポーネントを組み合わせてパイプライン全体を定義する。

from kfp import dsl, compiler

@dsl.pipeline(
    name="ml-training-pipeline",
    description="End-to-end ML training pipeline with evaluation and deployment",
)
def ml_training_pipeline(
    raw_data_path: str = "gs://my-bucket/data/raw.csv",
    test_size: float = 0.2,
    model_type: str = "random_forest",
    accuracy_threshold: float = 0.85,
):
    # Step 1: データ前処理
    preprocess_task = preprocess_data(
        raw_data_path=raw_data_path,
        test_size=test_size,
    )
    preprocess_task.set_cpu_limit("2")
    preprocess_task.set_memory_limit("4Gi")

    # Step 2: モデル学習
    train_task = train_model(
        train_dataset=preprocess_task.outputs["train_dataset"],
        model_type=model_type,
        hyperparameters={
            "n_estimators": 200,
            "max_depth": 10,
            "min_samples_split": 5,
        },
    )
    train_task.set_cpu_limit("4")
    train_task.set_memory_limit("8Gi")
    train_task.set_accelerator_type("nvidia.com/gpu")
    train_task.set_accelerator_limit(1)

    # Step 3: モデル評価
    eval_task = evaluate_model(
        test_dataset=preprocess_task.outputs["test_dataset"],
        trained_model=train_task.outputs["trained_model"],
        accuracy_threshold=accuracy_threshold,
    )

    # Step 4: 条件付きデプロイ (精度が閾値を超えた場合)
    with dsl.Condition(
        eval_task.outputs["deploy_decision"] == "approved",
        name="check-accuracy",
    ):
        deploy_task = deploy_model(
            model=train_task.outputs["trained_model"],
            serving_endpoint="ml-model-serving",
        )

# パイプラインのコンパイル
compiler.Compiler().compile(
    pipeline_func=ml_training_pipeline,
    package_path="ml_pipeline.yaml",
)

高度なパイプラインパターン

並列実行と条件分岐

@dsl.pipeline(name="parallel-training-pipeline")
def parallel_training_pipeline(
    raw_data_path: str,
    accuracy_threshold: float = 0.85,
):
    # データ前処理 (共通)
    preprocess_task = preprocess_data(
        raw_data_path=raw_data_path,
        test_size=0.2,
    )

    # 複数モデルの並列学習
    models = ["random_forest", "gradient_boosting", "xgboost"]
    train_tasks = []

    for model_type in models:
        train_task = train_model(
            train_dataset=preprocess_task.outputs["train_dataset"],
            model_type=model_type,
            hyperparameters={"n_estimators": 200, "max_depth": 10},
        )
        train_task.set_display_name(f"Train {model_type}")
        train_tasks.append(train_task)

    # 最適モデルの選択
    select_task = select_best_model(
        models=[t.outputs["trained_model"] for t in train_tasks],
        metrics=[t.outputs["metrics"] for t in train_tasks],
    )

    # チャンピオンモデルのデプロイ
    with dsl.Condition(
        select_task.outputs["best_accuracy"] >= accuracy_threshold,
        name="accuracy-gate",
    ):
        deploy_model(
            model=select_task.outputs["best_model"],
            serving_endpoint="champion-model",
        )

繰り返し実行と Exit Handler

@dsl.pipeline(name="robust-ml-pipeline")
def robust_ml_pipeline(raw_data_path: str):
    # Exit Handler: パイプラインの完了/失敗時に通知を送る
    notify_task = send_notification(
        pipeline_name="robust-ml-pipeline",
        notification_channel="slack",
    )

    with dsl.ExitHandler(exit_task=notify_task):
        # メインのパイプラインロジック
        preprocess_task = preprocess_data(
            raw_data_path=raw_data_path,
            test_size=0.2,
        )

        train_task = train_model(
            train_dataset=preprocess_task.outputs["train_dataset"],
            model_type="gradient_boosting",
            hyperparameters={"n_estimators": 300, "max_depth": 12},
        )

        eval_task = evaluate_model(
            test_dataset=preprocess_task.outputs["test_dataset"],
            trained_model=train_task.outputs["trained_model"],
            accuracy_threshold=0.85,
        )

# Recurring Run の設定 (KFP クライアント)
from kfp.client import Client

client = Client(host="https://kubeflow.example.com/pipeline")

# 毎日午前 2 時にパイプラインを実行
client.create_recurring_run(
    experiment_id="daily-training-exp",
    job_name="daily-model-retraining",
    pipeline_id="robust-ml-pipeline-v2",
    cron_expression="0 2 * * *",
    max_concurrency=1,
    params={
        "raw_data_path": "gs://my-bucket/data/daily/latest.csv",
    },
)

パイプラインのキャッシュとアーティファクト管理

キャッシュ戦略

KFP は、コンポーネントの入力が同一であれば以前の実行結果を再利用するキャッシュ機能に対応している。

# キャッシュの設定
@dsl.pipeline(name="cached-pipeline")
def cached_pipeline(raw_data_path: str):
    # キャッシュの有効化 (既定値: True)
    preprocess_task = preprocess_data(
        raw_data_path=raw_data_path,
        test_size=0.2,
    )
    preprocess_task.set_caching_options(enable_caching=True)

    # 学習ステップはキャッシュを無効化 (最新データで常に再学習)
    train_task = train_model(
        train_dataset=preprocess_task.outputs["train_dataset"],
        model_type="random_forest",
        hyperparameters={"n_estimators": 200},
    )
    train_task.set_caching_options(enable_caching=False)

アーティファクトの型と管理

from kfp.dsl import (
    Input, Output,
    Dataset, Model, Metrics,
    ClassificationMetrics, SlicedClassifications,
    Artifact, HTML, Markdown,
)

@dsl.component(base_image="python:3.11-slim")
def generate_evaluation_report(
    test_dataset: Input[Dataset],
    trained_model: Input[Model],
    classification_metrics: Output[ClassificationMetrics],
    html_report: Output[HTML],
    eval_metrics: Output[Metrics],
):
    """評価レポート生成コンポーネント"""
    import json

    # ClassificationMetrics: 混同行列の可視化
    classification_metrics.log_confusion_matrix(
        categories=["negative", "positive"],
        matrix=[[850, 50], [30, 270]],
    )

    # ROC カーブのロギング
    classification_metrics.log_roc_curve(
        fpr=[0.0, 0.1, 0.2, 0.5, 1.0],
        tpr=[0.0, 0.6, 0.8, 0.95, 1.0],
        threshold=[1.0, 0.8, 0.5, 0.2, 0.0],
    )

    # HTML レポートの生成
    report_content = "<h1>Model Evaluation Report</h1>"
    report_content += "<p>Accuracy: 0.933</p>"
    report_content += "<p>F1 Score: 0.891</p>"

    with open(html_report.path, "w") as f:
        f.write(report_content)

    # 数値メトリクス
    eval_metrics.log_metric("accuracy", 0.933)
    eval_metrics.log_metric("f1_score", 0.891)
    eval_metrics.log_metric("precision", 0.844)
    eval_metrics.log_metric("recall", 0.900)

ワークフローオーケストレーションツールの比較

ML ワークフローのオーケストレーションに使えるツールは複数ある。プロジェクトの要件に応じて適切なツールを選ぶ必要がある。

特性Kubeflow PipelinesApache AirflowArgo WorkflowsPrefect
主な用途ML パイプライン専用汎用データパイプライン汎用ワークフロー汎用データパイプライン
実行環境Kubernetes 必須多様な ExecutorKubernetes 必須ハイブリッド (サーバ/クラウド)
ML ネイティブ高い (MLMD, アーティファクト)低い (プラグインが必要)中程度中程度
UI/可視化ML 実験ダッシュボードDAG モニタリングワークフロー可視化フローダッシュボード
キャッシュコンポーネント単位タスク単位Memoizationタスク単位
スケーリングKubernetes ネイティブCelery/K8s ExecutorKubernetes ネイティブDask/Ray 統合
学習コスト高い中程度高い低い
コミュニティ活発 (CNCF)非常に活発 (Apache)活発 (CNCF)成長中
GPU 対応ネイティブ限定的ネイティブ外部統合が必要

選択の基準

Multi-Step ML パイプラインの実践例

パイプライン全体: データ準備からデプロイまで

from kfp import dsl, compiler
from kfp.dsl import Input, Output, Dataset, Model, Metrics

@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.1.4", "great-expectations==0.18.8"],
)
def validate_data(
    raw_data_path: str,
    validated_data: Output[Dataset],
    validation_metrics: Output[Metrics],
) -> str:
    """データ品質の検証"""
    import pandas as pd

    df = pd.read_csv(raw_data_path)

    # 基本的なデータ品質チェック
    checks = {
        "row_count_check": len(df) > 100,
        "null_ratio_check": df.isnull().mean().max() < 0.3,
        "duplicate_check": df.duplicated().mean() < 0.05,
        "target_balance_check": df["target"].value_counts(normalize=True).min() > 0.1,
    }

    all_passed = all(checks.values())

    for check_name, passed in checks.items():
        validation_metrics.log_metric(check_name, int(passed))

    validation_metrics.log_metric("total_rows", len(df))
    validation_metrics.log_metric("all_checks_passed", int(all_passed))

    if all_passed:
        df.to_csv(validated_data.path, index=False)
        return "passed"
    else:
        failed = [k for k, v in checks.items() if not v]
        raise ValueError(f"Data validation failed: {failed}")


@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.1.4", "scikit-learn==1.4.0"],
)
def feature_engineering(
    validated_data: Input[Dataset],
    feature_config: dict,
    features_dataset: Output[Dataset],
    feature_metrics: Output[Metrics],
):
    """フィーチャーエンジニアリング"""
    import pandas as pd
    import numpy as np
    from sklearn.preprocessing import StandardScaler, LabelEncoder

    df = pd.read_csv(validated_data.path)

    # 数値フィーチャーのスケーリング
    numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
    numeric_cols = [c for c in numeric_cols if c != "target"]

    scaler = StandardScaler()
    df[numeric_cols] = scaler.fit_transform(df[numeric_cols])

    # カテゴリフィーチャーのエンコーディング
    cat_cols = df.select_dtypes(include=["object"]).columns.tolist()
    for col in cat_cols:
        le = LabelEncoder()
        df[col] = le.fit_transform(df[col].astype(str))

    df.to_csv(features_dataset.path, index=False)

    feature_metrics.log_metric("numeric_features", len(numeric_cols))
    feature_metrics.log_metric("categorical_features", len(cat_cols))
    feature_metrics.log_metric("total_features", len(df.columns) - 1)


@dsl.component(
    base_image="gcr.io/my-project/ml-serving:v1.0",
)
def deploy_to_kserve(
    model: Input[Model],
    serving_endpoint: str,
    namespace: str,
) -> str:
    """KServe へモデルをデプロイ"""
    import subprocess
    import json
    import yaml

    inference_service = {
        "apiVersion": "serving.kserve.io/v1beta1",
        "kind": "InferenceService",
        "metadata": {
            "name": serving_endpoint,
            "namespace": namespace,
        },
        "spec": {
            "predictor": {
                "model": {
                    "modelFormat": {"name": "sklearn"},
                    "storageUri": model.uri,
                    "resources": {
                        "requests": {"cpu": "1", "memory": "2Gi"},
                        "limits": {"cpu": "2", "memory": "4Gi"},
                    },
                }
            }
        },
    }

    manifest_path = "/tmp/isvc.yaml"
    with open(manifest_path, "w") as f:
        yaml.dump(inference_service, f)

    result = subprocess.run(
        ["kubectl", "apply", "-f", manifest_path],
        capture_output=True, text=True,
    )

    if result.returncode != 0:
        raise RuntimeError(f"Deploy failed: {result.stderr}")

    return f"Deployed to {namespace}/{serving_endpoint}"


@dsl.pipeline(
    name="e2e-ml-pipeline",
    description="データ検証からモデルデプロイまでの ML パイプライン全体",
)
def e2e_ml_pipeline(
    raw_data_path: str = "gs://ml-data/raw/dataset.csv",
    model_type: str = "gradient_boosting",
    accuracy_threshold: float = 0.85,
    serving_endpoint: str = "fraud-detector",
    namespace: str = "ml-serving",
):
    # 通知のための Exit Handler
    notify = send_notification(
        pipeline_name="e2e-ml-pipeline",
        notification_channel="slack",
    )

    with dsl.ExitHandler(exit_task=notify):
        # 1. データ検証
        validate_task = validate_data(raw_data_path=raw_data_path)

        # 2. フィーチャーエンジニアリング
        feature_task = feature_engineering(
            validated_data=validate_task.outputs["validated_data"],
            feature_config={"scaling": "standard", "encoding": "label"},
        )

        # 3. データ分割
        split_task = preprocess_data(
            raw_data_path=feature_task.outputs["features_dataset"].uri,
            test_size=0.2,
        )

        # 4. モデル学習
        train_task = train_model(
            train_dataset=split_task.outputs["train_dataset"],
            model_type=model_type,
            hyperparameters={"n_estimators": 300, "max_depth": 12},
        )
        train_task.set_cpu_limit("4")
        train_task.set_memory_limit("16Gi")

        # 5. モデル評価
        eval_task = evaluate_model(
            test_dataset=split_task.outputs["test_dataset"],
            trained_model=train_task.outputs["trained_model"],
            accuracy_threshold=accuracy_threshold,
        )

        # 6. 条件付きデプロイ
        with dsl.Condition(
            eval_task.outputs["deploy_decision"] == "approved",
            name="deploy-gate",
        ):
            deploy_to_kserve(
                model=train_task.outputs["trained_model"],
                serving_endpoint=serving_endpoint,
                namespace=namespace,
            )

# コンパイルと投入
compiler.Compiler().compile(
    pipeline_func=e2e_ml_pipeline,
    package_path="e2e_ml_pipeline.yaml",
)

Kubernetes のリソース管理

Pod リソースとノードアフィニティの設定

@dsl.pipeline(name="resource-managed-pipeline")
def resource_managed_pipeline():
    train_task = train_model(
        train_dataset=preprocess_task.outputs["train_dataset"],
        model_type="xgboost",
        hyperparameters={"n_estimators": 500},
    )

    # リソース制限
    train_task.set_cpu_limit("8")
    train_task.set_memory_limit("32Gi")
    train_task.set_accelerator_type("nvidia.com/gpu")
    train_task.set_accelerator_limit(2)

    # ノードセレクタ (GPU ノードで実行)
    train_task.add_node_selector_constraint(
        label_name="cloud.google.com/gke-accelerator",
        value="nvidia-tesla-v100",
    )

    # Toleration の設定
    train_task.set_gpu_limit(2).add_toleration(
        key="nvidia.com/gpu",
        operator="Exists",
        effect="NoSchedule",
    )

    # PVC のマウント (大容量データ)
    train_task.add_pvolumes({
        "/mnt/data": dsl.PipelineVolume(
            pvc="ml-data-pvc",
            volume_name="data-volume",
        ),
    })

    # タイムアウトの設定 (秒単位)
    train_task.set_timeout(3600)  # 1 時間

    # リトライの設定
    train_task.set_retry(
        num_retries=3,
        policy="Always",
        backoff_duration="30s",
        backoff_factor=2.0,
        backoff_max_duration="600s",
    )

運用上の注意事項

リソース関連の注意点

  1. メモリ OOM: 大容量データセットを扱うコンポーネントには十分なメモリを割り当てる必要がある。Pandas の read_csv はデータサイズの 3-5 倍のメモリを消費する。
  2. GPU リソースの競合: 複数のパイプラインが同時に GPU を要求すると Pending 状態が長引く。ResourceQuota と PriorityClass を設定すること。
  3. PVC への同時アクセス: ReadWriteOnce の PVC は一つの Pod しかマウントできない。並列コンポーネントが同じ PVC にアクセスすると失敗する。

セキュリティ関連の注意点

  1. シークレット管理: パイプラインのパラメータに API キーやパスワードを直接渡してはいけない。Kubernetes Secret を環境変数としてマウントすること。
  2. イメージの脆弱性: ベースイメージのセキュリティ脆弱性を定期的にスキャンすること。python:3.11-slim の代わりに distroless イメージの利用を検討する。
  3. RBAC の設定: パイプラインのサービスアカウントに最小権限の原則を適用すること。

障害事例と復旧手順

事例 1: Pod OOMKilled

症状: コンポーネントの Pod が OOMKilled 状態で失敗する

# Pod の状態確認
kubectl get pods -n kubeflow -l pipeline/runid=run-abc123
kubectl describe pod train-model-xxxxx -n kubeflow

# イベントで OOMKilled を確認
# Last State: Terminated
#   Reason: OOMKilled
#   Exit Code: 137

復旧手順:

# 1. メモリ制限の引き上げ
train_task.set_memory_limit("64Gi")

# 2. データをチャンク単位で処理するようコンポーネントを修正
@dsl.component(base_image="python:3.11-slim")
def train_with_chunks(
    train_dataset: Input[Dataset],
    chunk_size: int,
    trained_model: Output[Model],
):
    import pandas as pd
    from sklearn.linear_model import SGDClassifier

    model = SGDClassifier(loss="log_loss")
    chunks = pd.read_csv(train_dataset.path, chunksize=chunk_size)

    for chunk in chunks:
        X = chunk.drop("target", axis=1)
        y = chunk["target"]
        model.partial_fit(X, y, classes=[0, 1])

    import joblib
    joblib.dump(model, trained_model.path)

事例 2: パイプラインのバージョン衝突

症状: パイプライン更新後に既存の Recurring Run が失敗する

復旧手順:

from kfp.client import Client

client = Client(host="https://kubeflow.example.com/pipeline")

# 1. 既存の Recurring Run を無効化
client.disable_recurring_run(recurring_run_id="run-xxx")

# 2. 新しいパイプラインバージョンをアップロード
pipeline_version = client.upload_pipeline_version(
    pipeline_package_path="ml_pipeline_v3.yaml",
    pipeline_version_name="v3.0",
    pipeline_id="ml-training-pipeline",
)

# 3. 新しい Recurring Run を作成
client.create_recurring_run(
    experiment_id="daily-training-exp",
    job_name="daily-model-retraining-v3",
    version_id=pipeline_version.pipeline_version_id,
    cron_expression="0 2 * * *",
    max_concurrency=1,
)

事例 3: メタデータ DB への接続失敗

症状: ML Metadata Service への接続エラーでアーティファクト追跡に失敗する

# MLMD サービスの状態確認
kubectl get pods -n kubeflow -l app=metadata-grpc-server
kubectl logs metadata-grpc-server-xxxxx -n kubeflow

# MySQL/PostgreSQL の接続確認
kubectl exec -it metadata-grpc-server-xxxxx -n kubeflow -- \
    mysql -h metadata-db -u root -p -e "SHOW DATABASES;"

# MLMD サービスの再起動
kubectl rollout restart deployment metadata-grpc-server -n kubeflow

プロダクションチェックリスト

インフラ設定

パイプライン開発

運用とモニタリング

セキュリティ

参考資料

コメント

まだコメントはありません。

ログインするとコメントできます