Spark の分散モデル学習

分散学習では、並列で動作する複数のコンピューティングリソースでモデルを学習します。Foundry は Spark 環境での分散モデル学習をサポートしています。分散学習では、次のことができます。

モデルをより高速かつ効率的に学習する： 分散学習では複数のノードを同時に活用し、モデル学習を高速化します。
学習データセットをスケールアップする： 学習タスクとデータをノード間で分割することで、単一ノードの方式ではメモリーに収まらない大規模なデータセットでもモデルを学習できます。

現在、Foundry の分散モデル学習で直接サポートされているのは、PySpark XGBoost 推定器のみです。今後、ほかの分散学習ライブラリーもサポートされる可能性があります。

以下の手順に沿って、Foundry で分散モデル学習を実行する方法を学びます。

1. Spark 環境を設定する

分散学習ライブラリーを使用する前に、Spark プロファイルを使用して Spark 環境を適切に設定する必要があります。

必須の Spark プロファイル

分散モデル学習を実行するには、以下のコード例のように、リポジトリーで KUBERNETES_OPEN_PORTS_ALL プロファイルをインポートして適用する必要があります。

Copied!1
2
3
4
5
6
7
8
from transforms.api import configure

@configure(profile=[
    "KUBERNETES_OPEN_PORTS_ALL",
])
@transform_df(...)
def compute(...):
    ...

繰り返しになりますが、分散学習では KUBERNETES_OPEN_PORTS_ALL プロファイルの適用が必須です。

2. 分散学習ライブラリーをセットアップする

XGBoost Spark 推定器をセットアップする

XGBoost は、SparkXGBClassifier ↗ または SparkXGBRegressor ↗ 推定器を介して、分散学習のためのシームレスな PySpark 統合を提供します。

SparkXGBClassifier の基本的なセットアップの例は次のとおりです。

Copied!1
2
3
4
5
6
7
8
9
10
from xgboost.spark import SparkXGBClassifier

@configure(profile=["KUBERNETES_OPEN_PORTS_ALL"])
@transform(...)
def compute(...):
    xgb = SparkXGBClassifier(
        features_col=<your_feature_col_name>,
        # 必要に応じてほかのパラメーターを指定します
    )
    model = xgb.fit(<your_training_dataframe>)

設定の詳細については、XGBoost Spark ドキュメント ↗を参照してください。

分散モデル学習の GPU サポートをセットアップする

分散モデル学習で GPU を活用するには、以下の手順に沿って進めます。

GPU をサポートするリソースキューにプロジェクトを追加します。
トランスフォームに EXECUTOR_GPU_ENABLED Spark プロファイルを追加して、GPU プロファイルを有効にします。
GPU の設定に応じて、device パラメーターを 'gpu' または 'cuda' に設定します。

例：

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
from transforms.api import configure

@configure(profile=[
    "KUBERNETES_OPEN_PORTS_ALL",
    "EXECUTOR_GPU_ENABLED",
])
@transform_df(...)
def compute(...):
    xgb = SparkXGBClassifier(
        ...,
        device='gpu'  # オプション: 'cpu', 'gpu', 'cuda'
    )
    model = xgb.fit(...)

GPU の設定について詳しくは、SparkXGBClassifier ドキュメント ↗を参照してください。

3. 分散ハイパーパラメーター最適化

必要に応じて、ハイパーパラメーター最適化を実行することもできます。トランスフォームでのハイパーパラメーター最適化には、Optuna ↗ の使用を推奨します。

Optuna は、追加のセットアップなしで Spark や分散学習ワークフローと良好に統合できます。
詳細については、Optuna ドキュメント ↗を参照してください。

4. モデルアダプターで SparkXGBClassifier または SparkXGBRegressor を使用する

palantir_models.serializers に組み込まれている XGBoostSerializer は、XGBoost の Spark 推定器（SparkXGBClassifier / SparkXGBRegressor）を直接サポートしていません。代わりに、次の手順を実行します。

通常どおり Spark XGBoost モデルを学習します。
.get_booster() を使用して、基になる Booster オブジェクトを抽出します。
モデルアダプターで Booster を使用します。

分散方式で推論を実行する必要がある場合でも、PySpark UDF 内で ModelAdapter を登録できます。詳細については、PySpark UDF を参照してください。

モデルアダプターの例

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from palantir_models.serializers import XGBoostSerializer
import palantir_models as pm

class BoosterModelAdapter(pm.ModelAdapter):

    @pm.auto_serialize(model=XGBoostSerializer())
    def __init__(self, booster):
        self.booster = booster

    @classmethod
    def api(cls):
        <...>

    def predict(self, input_df):
        return self.booster.predict(input_df)

学習とアダプターの初期化

Copied!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
from transforms.api import configure, transform
from transforms.api.schema import Input, ModelOutput
from xgboost.spark import SparkXGBClassifier  # または SparkXGBRegressor
from your_project.adapters import BoosterModelAdapter

@configure(profile=[
    "DYNAMIC_ALLOCATION_DISABLED",
    "NUM_EXECUTORS_2",
    "KUBERNETES_OPEN_PORTS_ALL"
])
@transform(
    training_df=Input(path="..."),
    model_output=ModelOutput(path="...")
)
def compute(ctx, training_df, model_output):
    # 1. SparkXGBClassifier を学習させます
    xgb = SparkXGBClassifier(...)
    spark_model = xgb.fit(training_df)

    # 2. 内部の Booster を抽出します
    booster = spark_model.get_booster()

    # 3. モデルアダプターを作成します
    adapter = BoosterModelAdapter(booster)

    # 4. モデルを Foundry に公開します
    model_output.publish(adapter)

アダプターの設定について詳しくは、モデルアダプターの概要を参照してください。

←

PREVIOUSSpark ML モデル

NEXTモデルアダプターのアップグレード（再訓練なし）

→