Spark ML

Spark 包含一個名為 MLlib 的機器學習函式庫。「Spark ML」通常用來指稱 MLlib 以 DataFrame 為基礎的 API，因為其 Java 與 Scala 類別位於 org.apache.spark.ml 下，而 Python 類別位於 pyspark.ml 下。

以下名稱容易混淆：

spark.mllib 並未移除，但新應用程式程式碼通常應使用 spark.ml。DataFrame API 與 Spark SQL 整合，具有一致的管線模型，並在 Java、Scala 與 Python 之間提供相似的 API。

Spark ML 提供分散式特徵準備、模型擬合、評估、超參數調校與批次推論。當訓練資料或評分 資料已經足夠龐大、能夠從 Spark 中獲益時，Spark ML 最為實用。它不是通用的深度學習 框架，也不會僅因為在叢集上執行，就自動讓每個演算法都變得有效率。

Spark ML 提供的功能 #

函式庫涵蓋數個類別：

• 分類： 邏輯斯迴歸、決策樹、隨機森林、梯度提升樹、樸素貝氏、線性支援向量機與 多層感知器
• 迴歸： 線性迴歸與廣義線性迴歸、決策樹、隨機森林、梯度提升樹、生存迴歸、保序 迴歸與因子分解機
• 分群： K-means、二分 K-means、高斯混合、潛在狄利克雷配置與冪次迭代分群
• 推薦： 交替最小平方法
• 頻繁模式： FP-growth 與 prefix-span
• 特徵工程： 分詞、類別編碼、特徵組合、縮放、雜湊、TF-IDF、離散化、插補、多項式 展開、降維與特徵選取
• 統計： 相關性、假設檢定與向量摘要
• 工作流程： 管線、評估器、交叉驗證、訓練驗證分割、參數網格與模型持久化

Java、Scala、Python 與 R 之間的演算法支援與參數有所不同。請查閱叢集所使用 Spark 版本的 API 文件，不要假設所有繫結都具有完全相同的涵蓋範圍。

作為 ML 資料集的 DataFrame #

Spark ML 消費並產生 DataFrame。監督式學習資料集通常包含以下結構：

+-----+-------------------+
|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： 儲存向量大小，以及非零索引與值。

對於獨熱編碼類別與雜湊文字等高維資料，稀疏向量不可或缺。將它們展開成密集陣列可能會 大幅增加記憶體與網路使用量。

預設欄位名稱只是慣例。大多數演算法允許呼叫者變更這些名稱：

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

其他常見輸出欄位包括：

• rawPrediction：未正規化的模型分數
• probability：模型提供時的類別機率
• prediction：選取的類別或迴歸值

DataFrame 綱要讓管線階段可以組合，但檢查會在執行階段進行。Spark 會在階段擬合或 轉換時，驗證必要欄位是否存在，以及型別是否相容。

Transformer 與 Estimator #

Spark ML 以兩種抽象概念統一演算法。

Transformer 將一個 DataFrame 轉換為另一個 DataFrame：

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

它通常會保留輸入欄位，同時附加新欄位。特徵 Transformer 與訓練完成的模型都是 Transformer。例如，Tokenizer 將文字欄位轉換為單字陣列，而 LogisticRegressionModel 將特徵向量轉換為預測欄位。

Estimator 從 DataFrame 學習並產生模型：

training DataFrame --fit()--> Model

Model 是 Transformer：

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

例如：

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。參數屬於特定的階段執行個體：

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

使用以下程式碼檢查支援的參數與預設值：

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

傳遞給 fit() 或 transform() 的 ParamMap 會覆寫先前在執行個體上設定的值：

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

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

由於參數繫結至階段執行個體，因此同一個參數網格中的兩個邏輯斯迴歸階段可以擁有 獨立的值。

特徵工程 #

大多數演算法預期一個數值向量欄位，但來源資料通常包含數值、類別、布林值與文字欄位。 特徵 Transformer 會將這些欄位轉換為可用的向量。

數值特徵 #

VectorAssembler 組合純量欄位與向量欄位：

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

缺失值必須以明確方式處理。Imputer 可以使用平均值、中位數或眾數取代缺失的數值。 它不會讓缺失在語意上變得無害；所選策略必須符合資料。

StandardScaler、MinMaxScaler 與 MaxAbsScaler 會重新縮放向量。對於以距離和梯度 為基礎的演算法，縮放十分重要；決策樹通常不需要縮放。

對稀疏向量進行中心化時必須小心。減去非零平均值會使隱含的零變成非零值，可能因此 讓稀疏資料集變成密集資料集。

類別特徵 #

StringIndexer 將類別字串轉換為數值索引：

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

OneHotEncoder 將索引轉換為稀疏指示向量：

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

Indexer 必須只在訓練資料上擬合。在測試資料或正式環境資料上獨立擬合，可能會為相同 類別指派不同索引。將兩個階段放入已擬合的 Pipeline，可保留訓練時的對應關係。

handleInvalid 控制收到 null 或未見過的值時的行為。無提示地選擇錯誤策略，可能導致 正式環境評分失敗，或掩蓋資料品質問題。

文字特徵 #

基本文字管線可以使用：

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

HashingTF 不需要學習詞彙表，但允許雜湊碰撞。CountVectorizer 會學習明確的詞彙表， 較容易檢查，但代價是必須儲存並分發詞彙表。

Spark ML 內建的文字特徵適合古典模型。它們不是現代語言模型分詞器、嵌入或面向 GPU 的 深度學習框架的替代品。

管線 #

Pipeline 將特徵準備與訓練記錄為單一有序工作流程：

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

Transformer 階段呼叫 transform()。Estimator 階段呼叫 fit() 產生 Model，接著由該 Model 轉換後續階段的資料。

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 都會在模型中成為已擬合的 Transformer。

這不只是語法上的便利。它確保驗證資料與正式環境資料使用從訓練資料學習的特徵對應。 在分割資料集前進行特徵擬合，會將驗證資料的資訊洩漏到訓練中，造成過度樂觀的評估結果。

Pipeline 階段雖以線性方式列出，但可以透過輸入欄位和輸出欄位形成 DAG。階段必須按照 拓撲順序排列，並且應使用唯一的執行個體，因為每個階段都有供參數系統使用的唯一 ID。

完整的訓練流程 #

可靠的監督式學習工作通常遵循以下順序：

1. 使用明確的綱要讀取原始資料。
2. 驗證標籤、範圍、null 處理與資料列識別性。
3. 將資料分割為訓練集、驗證集和最終測試集。
4. 僅使用訓練資料擬合完整的特徵與模型 Pipeline。
5. 使用訓練資料和驗證資料選取參數。
6. 僅在未使用的測試資料上評估選定模型一次。
7. 儲存已擬合的 PipelineModel 與其操作中繼資料。

例如：

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：

對於類別不平衡的分類，單獨使用準確率可能造成誤導。應根據業務成本檢查精確率、召回率、 F-measure、精確率-召回率曲線下面積、校準結果與混淆矩陣。排名良好的模型仍可能需要 另外選擇決策閾值。

RMSE、MAE 與 \(R^2\) 等迴歸指標回答不同問題。應根據錯誤成本選擇指標，而不是選擇看起來 最好的數字。

評估是分散式 Spark 工作，但某些摘要最終會將小型矩陣、歷史記錄或純量指標回傳至 driver。 不要僅為了計算 Spark 可以在叢集中彙總的指標，就收集整個預測 DataFrame。

超參數調校 #

Spark ML 提供兩個模型選取 Estimator：

• CrossValidator 在多個 fold 上評估每個參數對應。
• TrainValidationSplit 使用一次訓練/驗證分割評估每個參數對應，成本較低，但統計 穩健性較弱。

兩者都需要：

1. 一個 Estimator，通常是完整的 Pipeline
2. 一個參數對應陣列
3. 一個 Evaluator

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);

範例會進行以下訓練：

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

選取最佳參數後，交叉驗證會在完整輸入資料集上再次擬合該組態。因此，大型網格會消耗 大量 CPU、executor 記憶體、儲存 I/O 與時間。

parallelism 允許同時評估參數組態。它不會創造免費容量：過高的平行度可能讓多個訓練 工作競爭相同的 executor，導致效能變差。

如果特徵階段的參數也是問題的一部分，請調校完整 Pipeline。這也會在每個 fold 內個別 擬合具狀態的特徵 Transformer，從而避免驗證洩漏。

分散式執行 #

driver 建立 Pipeline 階段、分析 DataFrame 計畫、提交擬合與轉換工作，並保存產生的模型 中繼資料。executor 掃描分割區、計算特徵轉換、計算分散式統計量與梯度、建立部分樹統計量， 並對資料列進行評分。

確切的執行模式取決於演算法：

• 特徵轉換通常使用分割區上的一般 DataFrame 操作。
• 迭代式演算法會反覆執行工作並彙總更新。
• 樹演算法會交換直方圖與分割統計量。
• ALS 會將使用者因子與項目因子分割，並反覆交換必要的因子區塊。
• 部分已學習的模型狀態會被收集到 driver，之後分發以進行預測。

「分散式」不表示每個 executor 都獨立訓練一個模型物件。driver 協調單一擬合操作，而 executor 執行其分割區層級的工作。

模型大小仍然重要。係數、樹、詞彙表、類別對應與其他已擬合狀態可能需要駐留在 driver、 經過序列化，或傳送至 executor。資料集能夠放入叢集，並不表示任意大的模型也能放入。

如果重複擬合模型會重新讀取並重新計算訓練資料，請快取重複使用的訓練資料：

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

此動作會在調校開始前將快取具體化。在最後一次擬合後取消持久化：

training.unpersist();

快取每個中間 DataFrame 會浪費記憶體，並可能增加 spill。只持久化會由多個動作重複使用 的高成本資料。

儲存與載入模型 #

大多數 Spark ML Estimator、Model、Pipeline 與 PipelineModel 都實作 Spark 的 ML 持久化 API：

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

稍後載入：

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

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

儲存的 PipelineModel 包含已擬合的特徵階段與訓練完成的模型，但不會自動包含重現訓練 執行所需的所有操作事實。請額外儲存以下中繼資料：

• Spark 與應用程式版本
• 原始碼修訂版本
• 訓練資料快照或版本
• 特徵與標籤定義
• 分割方法與隨機種子
• 評估指標與選定閾值
• 參數對應
• 預期的輸入與輸出綱要

Spark 致力於維持小版本與修補版本之間的持久化相容性。跨主要 Spark 版本的相容性僅是 盡力而為，並不保證。升級正式環境推論工作前，請驗證載入與預測行為。

批次推論 #

最簡單的推論工作會載入模型並轉換 DataFrame：

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 driver 通常會產生高昂的操作成本。

線上服務可以採用以下方法：

• 在面向服務的執行環境中重現相容的模型；
• 將支援的模型類型匯出為可互通格式；
• 使用專用模型伺服器；或
• 在延遲需求允許時，以微批次處理請求評分。

並非每個 Pipeline 階段都有可移植的表示方式。在部署獨立的服務程式碼前，請測試離線 Spark 特徵與線上特徵之間的一致性。

自訂 Transformer 與 Estimator #

自訂 Transformer 可以將應用程式特定的特徵邏輯封裝成 Pipeline 階段。它必須定義參數、 輸入/輸出綱要行為、transform() 以及複製/持久化語意。

請優先使用原生 Spark SQL 運算式，而不是逐列 UDF：

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

原生運算式會參與 Catalyst 最佳化與程式碼生成。Java 或 Scala UDF 會引入不透明的呼叫 邊界；Python UDF 另外會跨越 JVM/Python 邊界，除非存在向量化執行路徑。

如果自訂狀態必須從資料中學習，請實作一個回傳相應 Model 的 Estimator。除非需要持久化、 調校與管線重複使用，否則一次性的 DataFrame 轉換序列可能比自訂階段更簡單。

Spark ML 與其他 ML 系統的比較 #

Spark ML 在以下情況下最具優勢：

• 訓練資料儲存在分散式資料湖中；
• 特徵準備需要大型聯結、彙總或掃描；
• 古典演算法能夠滿足建模需求；
• 批次訓練與推論符合產品需求；以及
• 組織已經能夠可靠地操作 Spark。

以下情況下，Spark ML 的吸引力較低：

• 準備好的資料集可以輕鬆放入單一機器；
• 需要以 GPU 為基礎的深度學習；
• 所需演算法尚未實作；
• 極低延遲的線上推論是主要工作負載；或
• 實驗迭代主要受叢集啟動與排程支配。

對於小型與中型表格式資料集，scikit-learn、XGBoost、LightGBM 或單機統計套件可能更 簡單且更快速。PyTorch 與 TensorFlow 等深度學習框架提供更豐富的神經網路與 GPU 支援。 Spark 仍可準備它們的資料集，但使用 Spark 進行特徵工程並不要求最終模型也使用 Spark ML。

常見錯誤 #

1. 新程式碼使用 spark.mllib。 除非維護既有 RDD 應用程式，否則優先使用以 DataFrame 為基礎的 spark.ml API。
2. 在分割資料前擬合特徵階段。 這會將資訊洩漏到驗證集與測試集。
3. 將訓練或預測資料收集到 driver。 將分散式資料保留為 DataFrame，並在 executor 上 進行彙總。
4. 從稀疏資料建立密集向量。 記憶體與 shuffle 大小可能大幅增加。
5. 忽略未見過的類別與 null。 透過階段參數與資料品質檢查定義明確的正式環境行為。
6. 為不平衡問題選擇準確率。 評估與業務成本一致的指標。
7. 啟動龐大的參數網格。 模型數量是參數值數量與 fold 數量的笛卡兒積。
8. 假設每個階段都能線性擴展。 模型狀態、driver 彙總、迭代、偏斜與 shuffle 仍可能 成為瓶頸。
9. 逐列記錄日誌或執行遠端呼叫。 這些副作用會在 executor 上平行執行，且難以安全重試。
10. 將儲存的模型視為完整的血緣資訊。 分別記錄資料、程式碼、參數、綱要與指標。

Spark ML 提供一種在分散式 DataFrame 上表達古典機器學習工作流程的一致方式。其核心抽象 不是單一演算法，而是同時攜帶特徵準備與預測能力的已擬合 PipelineModel。請在分散式 資料處理確實屬於問題的一部分時使用它，並在整個工作流程中留意評估、模型大小、服務 需求與資料洩漏。