spark mllib 分類預測之如何實現邏輯回歸

# Spark MLlib 分類預測之如何實現邏輯回歸

## 1. 引言

在大數據時代，機器學習已成為從海量數據中提取有價值信息的關鍵技術。Apache Spark作為領先的分布式計算框架，其內置的機器學習庫MLlib為開發者提供了豐富的算法實現。其中，邏輯回歸（Logistic Regression）作為經典的分類算法，因其模型簡單、解釋性強等特點，在金融風控、醫療診斷、廣告點擊率預測等領域廣泛應用。

本文將深入探討如何在Spark MLlib中實現邏輯回歸分類預測，包含以下核心內容：
- 邏輯回歸算法原理
- Spark MLlib環境準備
- 數據準備與特征工程
- 模型訓練與調優
- 模型評估與部署
- 完整代碼示例

## 2. 邏輯回歸算法原理

### 2.1 基本概念
邏輯回歸是一種用于解決二分類問題的統計方法，通過Sigmoid函數將線性回歸的輸出映射到(0,1)區間，表示屬于正類的概率：

$$
P(y=1|x) = \frac{1}{1+e^{-(w^Tx+b)}}
$$

其中：
- $w$為權重向量
- $b$為偏置項
- $x$為特征向量

### 2.2 Spark中的優化實現
Spark MLlib提供了兩種優化算法：
1. **L-BFGS**：擬牛頓法，適合特征維度適中的場景
2. **SGD**（隨機梯度下降）：適合大規模數據集

## 3. 環境準備

### 3.1 Spark環境配置
```python
from pyspark.sql import SparkSession
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.evaluation import BinaryClassificationEvaluator

spark = SparkSession.builder \
    .appName("LogisticRegressionExample") \
    .getOrCreate()

3.2 數據準備

以經典的鳶尾花數據集為例：

from sklearn.datasets import load_iris
import pandas as pd

# 加載數據
iris = load_iris()
df = pd.DataFrame(iris.data, columns=iris.feature_names)
df['label'] = (iris.target != 0).astype(int)  # 轉換為二分類問題

# 轉換為Spark DataFrame
spark_df = spark.createDataFrame(df)

4. 特征工程

4.1 特征向量化

assembler = VectorAssembler(
    inputCols=iris.feature_names,
    outputCol="features"
)
vectorized_df = assembler.transform(spark_df)

4.2 特征標準化（可選）

scaler = StandardScaler(
    inputCol="features",
    outputCol="scaledFeatures",
    withStd=True,
    withMean=True
)

scaler_model = scaler.fit(vectorized_df)
scaled_df = scaler_model.transform(vectorized_df)

4.3 數據劃分

train_df, test_df = scaled_df.randomSplit([0.7, 0.3], seed=42)

5. 模型訓練與調優

5.1 基礎模型訓練

lr = LogisticRegression(
    featuresCol="scaledFeatures",
    labelCol="label",
    maxIter=100,
    regParam=0.3,
    elasticNetParam=0.8
)

model = lr.fit(train_df)

5.2 超參數調優

使用CrossValidator進行網格搜索：

from pyspark.ml.tuning import ParamGridBuilder, CrossValidator

param_grid = (ParamGridBuilder()
    .addGrid(lr.regParam, [0.1, 0.3, 0.5])
    .addGrid(lr.elasticNetParam, [0.0, 0.5, 1.0])
    .addGrid(lr.maxIter, [50, 100])
    .build())

evaluator = BinaryClassificationEvaluator(
    labelCol="label",
    rawPredictionCol="rawPrediction",
    metricName="areaUnderROC"
)

cv = CrossValidator(
    estimator=lr,
    estimatorParamMaps=param_grid,
    evaluator=evaluator,
    numFolds=3
)

cv_model = cv.fit(train_df)
best_model = cv_model.bestModel

6. 模型評估

6.1 常用評估指標

from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# 在測試集上預測
predictions = best_model.transform(test_df)

# 計算準確率
accuracy = evaluator.evaluate(predictions)
print(f"Test Accuracy = {accuracy:.4f}")

# 計算其他指標
evaluator_multi = MulticlassClassificationEvaluator(
    labelCol="label",
    predictionCol="prediction",
    metricName="f1"
)
f1 = evaluator_multi.evaluate(predictions)
print(f"F1 Score = {f1:.4f}")

6.2 ROC曲線可視化

import matplotlib.pyplot as plt
from pyspark.mllib.evaluation import BinaryClassificationMetrics

# 轉換預測結果為RDD
prediction_and_labels = predictions.select("rawPrediction", "label").rdd.map(
    lambda row: (float(row['rawPrediction'][1]), float(row['label']))
)

# 計算ROC
metrics = BinaryClassificationMetrics(prediction_and_labels)
roc = metrics.roc().collect()

plt.figure(figsize=(8,6))
plt.plot([x[0] for x in roc], [x[1] for x in roc])
plt.plot([0, 1], [0, 1], 'r--')
plt.xlabel('False Positive Rate')
plt.ylabel('True Positive Rate')
plt.title('ROC Curve')
plt.show()

7. 模型部署與應用

7.1 模型保存與加載

# 保存模型
best_model.save("/path/to/model")

# 加載模型
from pyspark.ml.classification import LogisticRegressionModel
loaded_model = LogisticRegressionModel.load("/path/to/model")

7.2 實時預測示例

new_data = spark.createDataFrame([
    (5.1, 3.5, 1.4, 0.2),
    (6.2, 3.4, 5.4, 2.3)
], iris.feature_names)

# 相同的特征處理流程
new_vectorized = assembler.transform(new_data)
new_scaled = scaler_model.transform(new_vectorized)

# 預測
predictions = loaded_model.transform(new_scaled)
predictions.select("prediction").show()

8. 完整代碼示例

# 完整流程示例
from pyspark.sql import SparkSession
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
from sklearn.datasets import load_iris
import pandas as pd

# 初始化Spark
spark = SparkSession.builder \
    .appName("CompleteLRExample") \
    .getOrCreate()

# 數據準備
iris = load_iris()
df = pd.DataFrame(iris.data, columns=iris.feature_names)
df['label'] = (iris.target != 0).astype(int)
spark_df = spark.createDataFrame(df)

# 特征工程
assembler = VectorAssembler(
    inputCols=iris.feature_names,
    outputCol="features"
)
vectorized_df = assembler.transform(spark_df)

scaler = StandardScaler(
    inputCol="features",
    outputCol="scaledFeatures",
    withStd=True,
    withMean=True
)
scaler_model = scaler.fit(vectorized_df)
scaled_df = scaler_model.transform(vectorized_df)

# 數據劃分
train_df, test_df = scaled_df.randomSplit([0.7, 0.3], seed=42)

# 模型訓練與調優
lr = LogisticRegression(
    featuresCol="scaledFeatures",
    labelCol="label"
)

param_grid = (ParamGridBuilder()
    .addGrid(lr.regParam, [0.1, 0.3])
    .addGrid(lr.elasticNetParam, [0.0, 0.5])
    .build())

evaluator = BinaryClassificationEvaluator(
    labelCol="label",
    rawPredictionCol="rawPrediction",
    metricName="areaUnderROC"
)

cv = CrossValidator(
    estimator=lr,
    estimatorParamMaps=param_grid,
    evaluator=evaluator,
    numFolds=3
)

cv_model = cv.fit(train_df)
best_model = cv_model.bestModel

# 模型評估
predictions = best_model.transform(test_df)
accuracy = evaluator.evaluate(predictions)
print(f"Test AUC = {accuracy:.4f}")

# 模型保存
best_model.save("iris_lr_model")

9. 常見問題與解決方案

Q1: 如何處理類別不平衡數據？

• 使用classWeightCol參數指定類別權重
• 采用過采樣/欠采樣技術

Q2: 如何選擇正則化參數？

• 通過交叉驗證嘗試不同參數組合
• L1正則化(elasticNetParam=1.0)有助于特征選擇

Q3: 模型收斂速度慢怎么辦？

• 增加maxIter參數
• 調整stepSize參數（SGD優化器）
• 檢查特征是否需要標準化

10. 總結

本文詳細介紹了在Spark MLlib中實現邏輯回歸分類的完整流程，包括： 1. 環境配置與數據準備 2. 特征工程的最佳實踐 3. 模型訓練與超參數調優 4. 多種評估指標的應用 5. 模型部署的實用方法

Spark MLlib的邏輯回歸實現能夠高效處理大規模數據集，通過合理的參數調優和特征工程，可以構建出高性能的分類模型。建議讀者在實際應用中結合業務場景選擇合適的評估指標，并持續監控模型性能。

擴展閱讀： - Spark官方文檔 - MLlib指南 - 《Advanced Analytics with Spark》 - Sandy Ryza等著 - 梯度下降優化算法比較研究 “`