Spark ML

最終更新: 2026-08-06 17:32:00 +0800 +0800
tags: Spark | Machine-Learning | Mllib | Java
原文は英語で書かれています。あなたはAI翻訳版を読んでいます。

Spark には MLlib という機械学習ライブラリが含まれています。「Spark ML」という表現は、MLlib の DataFrame ベース API を指すために一般的に使われます。 これは、Java と Scala のクラスが org.apache.spark.ml の下にあり、Python のクラスが pyspark.ml の下にあるためです。

次の名称は混同しやすいものです。

名称意味
MLlibSpark の完全な機械学習ライブラリ
spark.ml主要な DataFrame ベースの MLlib API
spark.mllib以前の RDD ベース API。現在はメンテナンスモード
Spark MLspark.ml API の非公式名称

spark.mllib は削除されていませんが、新しいアプリケーションコードでは通常 spark.ml を使用すべきです。DataFrame API は Spark SQL と統合され、一貫した パイプラインモデルを持ち、Java、Scala、Python で類似した API を提供します。

Spark ML は、分散特徴量準備、モデル学習、評価、ハイパーパラメータ調整、バッチ推論を 提供します。学習データまたはスコアリングデータが Spark の恩恵を受けられるほど 大規模な場合に最も有用です。汎用的なディープラーニングフレームワークではなく、 単にクラスタ上で実行するだけですべてのアルゴリズムが効率的になるわけでもありません。

Spark ML が提供する機能 #

ライブラリはいくつかのカテゴリをカバーします。

  • 分類: ロジスティック回帰、決定木、ランダムフォレスト、勾配ブースティング木、 ナイーブベイズ、線形サポートベクターマシン、多層パーセプトロン
  • 回帰: 線形回帰および一般化線形回帰、決定木、ランダムフォレスト、 勾配ブースティング木、生存回帰、等張回帰、因子分解マシン
  • クラスタリング: K-means、二分 K-means、ガウス混合、潜在ディリクレ配分、 べき乗反復クラスタリング
  • 推薦: 交互最小二乗法
  • 頻出パターン: FP-growth、prefix-span
  • 特徴量エンジニアリング: トークン化、カテゴリエンコーディング、特徴量の結合、 スケーリング、ハッシュ化、TF-IDF、離散化、補完、多項式展開、次元削減、特徴量選択
  • 統計: 相関、仮説検定、ベクトル要約
  • ワークフロー: パイプライン、評価器、交差検証、学習・検証分割、 パラメータグリッド、モデル永続化

アルゴリズムのサポートとパラメータは Java、Scala、Python、R の間で異なります。 すべてのバインディングが同一の範囲を提供すると仮定せず、クラスタで使用している Spark バージョンの API ドキュメントを確認してください。

ML データセットとしての DataFrame #

Spark ML は DataFrame を入力および出力します。教師あり学習用のデータセットには、 慣例として次のような構造を持たせます。

text 
+-----+-------------------+
|label|features           |
+-----+-------------------+
|1.0  |[2.1, 0.0, 3.4]    |
|0.0  |(3,[0,2],[1.2,4.1])|
+-----+-------------------+

label は通常 Double であり、features は org.apache.spark.ml.linalg.Vector です。ベクトルには次の種類があります。

  • Dense: すべての値を格納します。
  • Sparse: ベクトルサイズと、ゼロでないインデックスおよび値だけを格納します。

スパースベクトルは、ワンホットエンコードされたカテゴリやハッシュ化されたテキストなど、 高次元データに不可欠です。これを dense 配列に展開すると、メモリ使用量とネットワーク 使用量が大幅に増加する可能性があります。

デフォルトの列名は単なる規約です。多くのアルゴリズムでは変更できます。

java 
new LogisticRegression()
    .setLabelCol("isFraud")
    .setFeaturesCol("modelFeatures")
    .setPredictionCol("predictedFraud");

その他の一般的な出力列は次のとおりです。

  • rawPrediction: 正規化されていないモデルスコア
  • probability: モデルが提供する場合のクラス確率
  • prediction: 選択されたクラスまたは回帰値

DataFrame のスキーマによりパイプラインステージを組み合わせられますが、検査は 実行時に行われます。Spark は、ステージの学習時または変換時に、必要な列が存在し、 互換性のある型であることを検証します。

Transformer と Estimator #

Spark ML は、アルゴリズムを 2 つの抽象化に統一しています。

Transformer は、ある DataFrame を別の DataFrame に変換します。

text 
input DataFrame --transform()--> output DataFrame

通常は入力列を保持したまま列を追加します。特徴量 Transformer と学習済みモデルは どちらも Transformer です。たとえば、Tokenizer はテキスト列を単語配列に変換し、 LogisticRegressionModel は特徴量ベクトルを予測列に変換します。

Estimator は DataFrame から学習し、モデルを生成します。

text 
training DataFrame --fit()--> Model

Model は Transformer です。

text 
test DataFrame --model.transform()--> predictions

例を示します。

java 
LogisticRegression trainer = new LogisticRegression()
    .setMaxIter(50)
    .setRegParam(0.01);

LogisticRegressionModel model = trainer.fit(training);
Dataset<Row> predictions = model.transform(test);

fit() と transform() は、分散 DataFrame 処理を記述するオブジェクトを返します。 Spark SQL は可能な限り遅延評価されます。write、evaluate、count、collect などの アクションによってジョブが実行されます。

パラメータ #

すべての Estimator と Transformer は共通のパラメータ API を実装しています。 パラメータは特定のステージインスタンスに属します。

java 
LogisticRegression lr = new LogisticRegression()
    .setMaxIter(50)
    .setRegParam(0.01)
    .setElasticNetParam(0.5);

サポートされているパラメータとデフォルト値は、次のように確認できます。

java 
System.out.println(lr.explainParams());
System.out.println(lr.extractParamMap());

fit() または transform() に渡した ParamMap は、インスタンスにあらかじめ設定された 値を上書きします。

java 
ParamMap override = new ParamMap()
    .put(lr.maxIter(), 100)
    .put(lr.regParam(), 0.05);

LogisticRegressionModel model = lr.fit(training, override);

パラメータはステージインスタンスに束縛されるため、1 つのパラメータグリッド内で 2 つのロジスティック回帰ステージに独立した値を設定できます。

特徴量エンジニアリング #

多くのアルゴリズムは 1 つの数値ベクトル列を想定しますが、入力データには通常、 数値、カテゴリ、真偽値、テキストの列が含まれます。特徴量 Transformer は、これらの列を 利用可能なベクトルに変換します。

数値特徴量 #

VectorAssembler はスカラー列とベクトル列を結合します。

java 
VectorAssembler assembler = new VectorAssembler()
    .setInputCols(new String[] {"age", "income", "accountAgeDays"})
    .setOutputCol("rawFeatures");

欠損値は意図的に処理する必要があります。Imputer は、平均値、中央値、最頻値を 用いて欠損した数値を置換できます。欠損値の意味上の影響をなくすものではないため、 戦略はデータに適合していなければなりません。

StandardScaler、MinMaxScaler、MaxAbsScaler はベクトルを再スケーリングします。 距離ベースおよび勾配ベースのアルゴリズムではスケーリングが重要ですが、決定木では 通常必要ありません。

スパースベクトルのセンタリングには注意してください。ゼロでない平均を減算すると 暗黙のゼロがゼロでなくなり、スパースデータセットが dense になる可能性があります。

カテゴリ特徴量 #

StringIndexer はカテゴリ文字列を数値インデックスに変換します。

java 
StringIndexer countryIndexer = new StringIndexer()
    .setInputCol("country")
    .setOutputCol("countryIndex")
    .setHandleInvalid("keep");

OneHotEncoder はインデックスをスパースな指示ベクトルに変換します。

java 
OneHotEncoder countryEncoder = new OneHotEncoder()
    .setInputCol("countryIndex")
    .setOutputCol("countryVector");

Indexer は学習データだけで学習させる必要があります。テストデータや本番データで 個別に学習すると、同じカテゴリに異なるインデックスが割り当てられる可能性があります。 両方のステージを学習済み Pipeline に含めることで、学習時のマッピングを維持できます。

handleInvalid は、null または未知の値が到着したときの動作を制御します。誤った ポリシーを暗黙に選択すると、本番スコアリングを失敗させるか、データ品質問題を 隠蔽する可能性があります。

テキスト特徴量 #

基本的なテキストパイプラインは次のように構成できます。

text 
text
  -> RegexTokenizer
  -> StopWordsRemover
  -> HashingTF or CountVectorizer
  -> IDF
  -> feature vector

HashingTF は語彙の学習を不要にしますが、ハッシュ衝突が発生する可能性があります。 CountVectorizer は明示的な語彙を学習するため検査しやすい一方、語彙の保存と配布が 必要になります。

Spark ML の組み込みテキスト特徴量は、古典的モデルに有用です。現代の言語モデル用 トークナイザー、埋め込み、GPU 指向のディープラーニングフレームワークの代替では ありません。

Pipeline #

Pipeline は特徴量準備と学習を 1 つの順序付きワークフローとして記録します。

text 
raw data
  -> StringIndexer
  -> OneHotEncoder
  -> VectorAssembler
  -> StandardScaler
  -> LogisticRegression

Transformer ステージは transform() を呼び出します。Estimator ステージは fit() を 呼び出して Model を生成し、その Model が後続ステージのデータを変換します。

java 
StringIndexer indexer = new StringIndexer()
    .setInputCol("country")
    .setOutputCol("countryIndex")
    .setHandleInvalid("keep");

OneHotEncoder encoder = new OneHotEncoder()
    .setInputCol("countryIndex")
    .setOutputCol("countryVector");

VectorAssembler assembler = new VectorAssembler()
    .setInputCols(new String[] {"age", "income", "countryVector"})
    .setOutputCol("features");

LogisticRegression classifier = new LogisticRegression()
    .setLabelCol("label")
    .setFeaturesCol("features");

Pipeline pipeline = new Pipeline().setStages(new PipelineStage[] {
    indexer,
    encoder,
    assembler,
    classifier
});

PipelineModel model = pipeline.fit(training);
Dataset<Row> predictions = model.transform(test);

Pipeline.fit() 自体が Estimator 操作であり、Transformer である PipelineModel を 返します。元の Pipeline に含まれるすべての Estimator は、Model 内で学習済み Transformer になります。

これは単なる構文上の利便性ではありません。検証データと本番データが、学習データから 学習した特徴量マッピングを使用することを保証します。データセットを分割する前に 特徴量の学習を行うと、検証データの情報が学習に漏洩し、評価が楽観的になります。

Pipeline ステージは線形に列挙されますが、入力列と出力列を介して DAG を形成する 場合があります。ステージはトポロジカル順序で並べる必要があり、各ステージは パラメータシステムで使用される一意の ID を持つため、一意なインスタンスにすべきです。

完全な学習フロー #

信頼性のある教師あり学習ジョブは、次のような順序に従います。

  1. 明示的なスキーマで生データを読み込む。
  2. ラベル、範囲、null 処理、行の識別性を検証する。
  3. データを学習、検証、最終テストの各セットに分割する。
  4. 特徴量処理とモデルを含む Pipeline 全体を学習データだけで学習させる。
  5. 学習データと検証データを使ってパラメータを選択する。
  6. 選択したモデルを未使用のテストデータで一度だけ評価する。
  7. 学習済み PipelineModel と運用メタデータを保存する。

例を示します。

java 
Dataset<Row>[] split = input.randomSplit(
    new double[] {0.8, 0.2},
    12345L
);

Dataset<Row> training = split[0].cache();
Dataset<Row> test = split[1];

PipelineModel model = pipeline.fit(training);
Dataset<Row> predictions = model.transform(test);

BinaryClassificationEvaluator evaluator =
    new BinaryClassificationEvaluator()
        .setLabelCol("label")
        .setRawPredictionCol("rawPrediction")
        .setMetricName("areaUnderROC");

double areaUnderRoc = evaluator.evaluate(predictions);

API が対応している場合はシードを設定してください。ただし、分散ジョブがビット単位で 決定論的になるとは限りません。パーティション分割、浮動小数点数の集約順序、入力 ファイルの配置、アルゴリズムの実装によって結果が変わる可能性があります。

時間依存データでは、ランダム分割は通常誤りです。未来から学習しないよう、より早い イベントで学習し、より後のイベントで検証してください。グループ化されたエンティティ では、同一ユーザーまたはデバイスのイベントが両側に現れないよう、グループを考慮した 分割が必要になる場合もあります。

評価 #

タスクに合った Evaluator を選択してください。

タスク一般的な Evaluator
二値分類BinaryClassificationEvaluator
多クラス分類MulticlassClassificationEvaluator
マルチラベル分類MultilabelClassificationEvaluator
回帰RegressionEvaluator
ランキングRankingEvaluator
クラスタリングClusteringEvaluator

不均衡な分類では、正解率だけでは誤解を招く可能性があります。ビジネス上のコストに 応じて、適合率、再現率、F 値、適合率-再現率曲線下面積、キャリブレーション、混同行列を 確認してください。ランキング性能が高いモデルでも、決定しきい値を別途選択する必要が ある場合があります。

RMSE、MAE、\(R^2\) などの回帰指標は、それぞれ異なる問いに答えます。最も良く見える 数値ではなく、誤差のコストに基づいて指標を選択してください。

評価は分散 Spark ジョブですが、一部の要約は最終的に小さな行列、履歴、スカラー指標を ドライバーに返します。Spark がクラスタ内で集約できる指標を計算するためだけに、 予測 DataFrame 全体を収集することは避けてください。

ハイパーパラメータ調整 #

Spark ML は、モデル選択用に 2 つの Estimator を提供します。

  • CrossValidator は、複数の fold にわたって各パラメータマップを評価します。
  • TrainValidationSplit は、1 つの学習・検証分割で各パラメータマップを評価します。 安価ですが、統計的な頑健性は低くなります。

どちらも次の要素を必要とします。

  1. Estimator。完全な Pipeline が使われることが多い。
  2. パラメータマップの配列。
  3. Evaluator。
java 
ParamMap[] grid = new ParamGridBuilder()
    .addGrid(classifier.regParam(), new double[] {0.001, 0.01, 0.1})
    .addGrid(classifier.elasticNetParam(), new double[] {0.0, 0.5, 1.0})
    .build();

CrossValidator cv = new CrossValidator()
    .setEstimator(pipeline)
    .setEvaluator(new BinaryClassificationEvaluator())
    .setEstimatorParamMaps(grid)
    .setNumFolds(3)
    .setParallelism(3)
    .setSeed(12345L);

CrossValidatorModel selectedModel = cv.fit(training);

この例では次の学習を行います。

text 
3 regParam values
x 3 elasticNetParam values
x 3 folds
= 27 model fits

最良のパラメータを選択した後、交差検証はその構成を入力データ全体でもう一度学習します。 したがって、大規模なグリッドは大量の CPU、Executor メモリ、ストレージ I/O、時間を 消費します。

parallelism により、パラメータ構成を同時に評価できます。これは無償の容量を作るもの ではありません。過剰な並列性は、複数の学習ジョブを同じ Executor で競合させ、性能を 低下させる可能性があります。

特徴量ステージのパラメータも問題の対象である場合は、完全な Pipeline を調整してください。 これにより、状態を持つ特徴量 Transformer も各 fold 内で個別に学習され、検証データの 漏洩を防げます。

分散実行 #

ドライバーは Pipeline ステージの作成、DataFrame プランの解析、学習および変換ジョブの 送信、生成されたモデルメタデータの保持を行います。Executor はパーティションを走査し、 特徴量変換を計算し、分散統計量と勾配を計算し、部分的な木の統計量を構築し、行を スコアリングします。

正確な実行パターンはアルゴリズムによって異なります。

  • 特徴量変換は、パーティション上の通常の DataFrame 操作を使用することが多い。
  • 反復アルゴリズムはジョブを繰り返し実行し、更新を集約する。
  • 木アルゴリズムはヒストグラムと分割統計量を交換する。
  • ALS はユーザー因子とアイテム因子をパーティション分割し、必要な因子ブロックを 繰り返し交換する。
  • 学習済みモデルの状態の一部はドライバーに収集され、後で予測のために配布される 場合がある。

「分散」とは、各 Executor で 1 つのモデルオブジェクトを個別に学習するという意味では ありません。ドライバーが 1 つの学習処理を調整し、Executor がそのパーティション単位の 作業を実行します。

モデルサイズも重要です。係数、木、語彙、カテゴリマップなどの学習済み状態は、 ドライバー上に置く、シリアライズする、Executor に送信する必要がある場合があります。 データセットがクラスタに収まることは、任意の大きさのモデルが収まることを意味しません。

繰り返しのモデル学習でデータの再読み込みと再計算が発生する場合は、再利用する学習 データをキャッシュしてください。

java 
Dataset<Row> training = preparedTraining.cache();
training.count();

このアクションにより、調整開始前にキャッシュが具体化されます。最後の学習後に キャッシュを解除します。

java 
training.unpersist();

すべての中間 DataFrame をキャッシュすると、メモリを浪費し、spill が増える可能性があります。 複数のアクションで再利用される高コストなデータだけを永続化してください。

モデルの保存と読み込み #

ほとんどの Spark ML Estimator、Model、Pipeline、PipelineModel は、Spark の ML 永続化 API を実装しています。

java 
model.write()
    .overwrite()
    .save("hdfs:///models/fraud/2026-07-20");

後で読み込むには次のようにします。

java 
PipelineModel model = PipelineModel.load(
    "hdfs:///models/fraud/2026-07-20"
);

Dataset<Row> predictions = model.transform(input);

保存された PipelineModel には、学習済みの特徴量ステージと学習済みモデルが含まれます。 ただし、学習実行を再現するために必要なすべての運用情報が自動的に含まれるわけでは ありません。次のような追加メタデータを保存してください。

  • Spark とアプリケーションのバージョン
  • ソースコードのリビジョン
  • 学習データのスナップショットまたはバージョン
  • 特徴量とラベルの定義
  • 分割方法と乱数シード
  • 評価指標と選択したしきい値
  • パラメータマップ
  • 想定する入力スキーマと出力スキーマ

Spark はマイナーバージョンおよびパッチバージョン間の永続化互換性を目指しています。 メジャーバージョン間の互換性はベストエフォートであり、保証されません。本番推論 ジョブをアップグレードする前に、読み込みと予測の動作を検証してください。

バッチ推論 #

最も単純な推論ジョブは、モデルを読み込み、DataFrame を変換します。

java 
PipelineModel model = PipelineModel.load(modelPath);

Dataset<Row> predictions = model.transform(input)
    .select("recordId", "probability", "prediction");

predictions.write()
    .mode(SaveMode.Overwrite)
    .parquet(outputPath);

変換は遅延評価されます。出力アクションが実行されると、Executor がスコアリングを行います。

Spark ML はバッチ推論に自然に適しています。低レイテンシのオンラインサービングは 別のワークロードです。リクエストごとに Spark ジョブを起動するのは不適切であり、 すべての Web サービスインスタンスに Spark ドライバーを組み込むと、通常は運用コストが 高くなります。

オンラインサービングでは、次のような方法が考えられます。

  • サービング指向のランタイムで互換性のあるモデルを再現する。
  • 対応するモデル形式を相互運用可能な形式にエクスポートする。
  • 専用のモデルサーバーを使用する。
  • レイテンシ要件が許容する場合、リクエストをマイクロバッチでスコアリングする。

すべての Pipeline ステージに移植可能な表現があるわけではありません。別個のサービング コードをデプロイする前に、オフライン Spark 特徴量とオンライン特徴量の一致をテスト してください。

カスタム Transformer と Estimator #

カスタム Transformer は、アプリケーション固有の特徴量ロジックを Pipeline ステージ としてパッケージ化できます。パラメータ、入力・出力スキーマの動作、transform()、 コピーおよび永続化のセマンティクスを定義する必要があります。

行ごとの UDF よりも、ネイティブ Spark SQL 式を優先してください。

java 
Dataset<Row> prepared = input.withColumn(
    "amountLog",
    functions.log1p(functions.col("amount"))
);

ネイティブ式は Catalyst の最適化とコード生成に参加します。Java または Scala の UDF は 不透明な呼び出し境界を導入します。Python UDF は、ベクトル化された実行パスが利用 できない限り、JVM と Python の境界も越えます。

データから学習する必要があるカスタム状態がある場合は、対応する Model を返す Estimator を実装してください。永続化、調整、Pipeline での再利用が必要でない限り、1 回限りの DataFrame 変換の連続処理の方がカスタムステージより単純な場合があります。

Spark ML と他の ML システムの比較 #

Spark ML は次の条件で強みを発揮します。

  • 学習データが分散データレイクに保存されている。
  • 特徴量準備に大規模な結合、集約、スキャンが必要である。
  • 古典的アルゴリズムでモデリング要件を満たせる。
  • バッチ学習とバッチ推論がプロダクトに適している。
  • 組織がすでに Spark を信頼性高く運用している。

次の条件では魅力が低下します。

  • 準備済みデータセットが 1 台のマシンに十分収まる。
  • GPU ベースのディープラーニングが必要である。
  • 必要なアルゴリズムが実装されていない。
  • 非常に低レイテンシのオンライン推論が主なワークロードである。
  • 実験の反復がクラスタ起動とスケジューリングに支配される。

小規模および中規模の表形式データセットでは、scikit-learn、XGBoost、LightGBM、 または単一マシン向けの統計パッケージの方が単純で高速な場合があります。PyTorch や TensorFlow などのディープラーニングフレームワークは、より豊富なニューラルネットワーク および GPU サポートを提供します。Spark でそれらのデータセットを準備することは可能 ですが、特徴量エンジニアリングに Spark を使用することは、最終モデルに Spark ML を 使用することを要求しません。

よくある間違い #

  1. 新しいコードで spark.mllib を使用する。 既存の RDD アプリケーションを保守する 場合を除き、DataFrame ベースの spark.ml API を優先してください。
  2. データ分割前に特徴量ステージを学習する。 検証セットとテストセットの情報が漏洩します。
  3. 学習データまたは予測データをドライバーに収集する。 分散データは DataFrame として 保持し、Executor 上で集約してください。
  4. スパースデータから dense ベクトルを作成する。 メモリ使用量と shuffle サイズが 大幅に増加する可能性があります。
  5. 未知のカテゴリと null を無視する。 ステージパラメータとデータ品質チェックで、 本番時の動作を明示してください。
  6. 不均衡な問題で正解率を選択する。 ビジネスコストに対応する指標を評価してください。
  7. 巨大なパラメータグリッドを起動する。 モデル数はパラメータ値と fold 数の直積です。
  8. すべてのステージが線形にスケールすると仮定する。 モデル状態、ドライバーでの集約、 反復、偏り、shuffle がボトルネックとして残る可能性があります。
  9. 行ごとにログ出力またはリモート呼び出しを行う。 こうした副作用は Executor 上で 並列に実行され、安全な再試行が困難です。
  10. 保存済みモデルを完全な系譜情報として扱う。 データ、コード、パラメータ、スキーマ、 指標を別途記録してください。

Spark ML は、分散 DataFrame 上で古典的な機械学習ワークフローを表現する一貫した方法を 提供します。中心となる抽象化は個々のアルゴリズムではなく、特徴量準備と予測をまとめて 保持する学習済み PipelineModel です。分散データ処理が問題の一部である場合に使用し、 ワークフロー全体を通じて評価、モデルサイズ、サービング要件、データ漏洩を考慮してください。

コメント

GitHub で閲覧・コメントする（投稿にはログインが必要です）