Spark機器學習實戰(zhàn)（五）用分類模型判別頁面內(nèi)容是否長期有效

這篇文章討論的是分類模型，完成的任務是判別一篇文章的內(nèi)容是否長久有效。比如，新聞就不具有長久有效的特質(zhì)，三個月前的新聞沒有什么價值，而科普文章則有。我們將會利用Spark的MLlib構(gòu)建邏輯回歸，SVM，樸素貝葉斯以及決策樹模型來對同一個數(shù)據(jù)集進行訓練。以一定標準來評價模型并介紹調(diào)優(yōu)的方法。

文章中列出了關鍵代碼，完整代碼見我的github repository，這篇文章的代碼在chapter05/src/main/scala/ScalaApp.scala

第1步：準備訓練數(shù)據(jù)

這次要訓練的數(shù)據(jù)來自于Kaggle，任務如上所述，我們把其中的train.tsv文件下載下來，作為我們的訓練集。我們先來查看一下我們下載下來的數(shù)據(jù)大概是什么樣子的。我截取了其中某一條數(shù)據(jù)。

"http://www.bloomberg.com/news/2010-12-23/ibm-predicts-holographic-calls-air-breathing-batteries-by-2015.html"  
"4042"  ".........."    "business"  "0.789131"  "2.055555556"   "0.676470588"   "0.205882353"   
"0.047058824"   "0.023529412"   "0.443783175"   "0" "0" "0.09077381"    "0" "0.245831182"   
"0.003883495"   "1" "1" "24"    "0" "5424"  "170"   "8" "0.152941176"   "0.079129575"   "0"

嗯看起來很混亂，其實并不復雜，每條數(shù)據(jù)由tab隔開。內(nèi)容順序依次為：url，urlid，頁面內(nèi)容，內(nèi)容分類，若干數(shù)值特征，最后是0或1表示的內(nèi)容長久與否，即標簽。

我們首先用這條shell命令把數(shù)據(jù)的第一行去除掉。

$ sed 1d train.tsv > train_noheader.tsv

Spark的分類模型訓練數(shù)據(jù)是以類LabeledPoint表示的，非常容易理解。我們構(gòu)建該類組成的RDD就算是準備好訓練數(shù)據(jù)了。其中有些數(shù)據(jù)是缺失的，用問號表示，我們把它替換成0。而樸素貝葉斯只接受非零輸入，我們簡單地把負數(shù)也都替換成0。url和urlid不能作為特征。文本特征很分類特征又有點麻煩，所以我們現(xiàn)在只截取了數(shù)值特征作為訓練輸入，標簽在最后。

    val sc: SparkContext = new SparkContext("local[2]", "First Spark App")
    sc.setLogLevel("ERROR")
    val rawData = sc.textFile("data/train_noheader.tsv")
    val records = rawData.map(line => line.split("\t"))
    val data = records.map { r =>
      val trimmed = r.map(_.replaceAll("\"", ""))
      val label = trimmed(r.size - 1).toInt
      val features = trimmed.slice(4, r.size - 1).map(d => if (d == "?") 0.0 else d.toDouble)
      LabeledPoint(label, Vectors.dense(features))
    }
    data.cache()
    val numData = data.count
    val nbData = records.map { r =>
      val trimmed = r.map(_.replaceAll("\"", ""))
      val label = trimmed(r.size - 1).toInt
      val features = trimmed.slice(4, r.size - 1).map(d => if (d == "?") 0.0 else d.toDouble)
        .map(d => if (d < 0) 0.0 else d)
      LabeledPoint(label, Vectors.dense(features))
    }

第2步：訓練分類模型

模型的構(gòu)建在Spark中異常簡單，import一些類調(diào)用一些API，參數(shù)都選默認，告知訓練迭代次數(shù)即可。

    val numIterations = 10
    val maxTreeDepth = 5
    val lrModel = LogisticRegressionWithSGD.train(data, numIterations)
    val svmModel = SVMWithSGD.train(data, numIterations)
    val nbModel = NaiveBayes.train(nbData)
    val dtModel = DecisionTree.train(data, Algo.Classification, Entropy, maxTreeDepth)

第3步：評價分類模型

評價分類模型我們采用以下三種標準：

正確率

很簡單，正確數(shù)/總數(shù)

    val lrTotalCorrect = data.map { lp =>
      if (lrModel.predict(lp.features) == lp.label) 1 else 0}.sum()
    val lrAccuracy = lrTotalCorrect / data.count
    println("lrAccuracy:" + lrAccuracy)
    val svmTotalCorrect = data.map { lp =>
      if (svmModel.predict(lp.features) == lp.label) 1 else 0}.sum()
    val svmAccuracy = svmTotalCorrect / data.count
    println("svmAccuracy:" + svmAccuracy)
    val nbTotalCorrect = nbData.map { lp =>
      if (nbModel.predict(lp.features) == lp.label) 1 else 0}.sum()
    val nbAccuracy = nbTotalCorrect / data.count
    println("nbAccuracy:" + nbAccuracy)
    val dtTotalCorrect = data.map { lp =>
      val score = dtModel.predict(lp.features)
      val predicted = if (score > 0.5) 1 else 0
      if (predicted == lp.label) 1 else 0}.sum()
    val dtAccuracy = dtTotalCorrect / data.count
    println("dtAccuracy:" + dtAccuracy)

結(jié)果如下：

lrAccuracy:0.5146720757268425
svmAccuracy:0.5146720757268425
nbAccuracy:0.5803921568627451
dtAccuracy:0.6482758620689655

準確率（precision）和召回率（recall）

準確率即 - 被你判為真的判對了多少？真陽/(真陽+假陽)

召回率即 - 真的被你判出來了多少？真陽/(真陽+假陰)

準確率和召回率受到判決閾值的影響，一般分類模型的輸出為0～1之間的一個數(shù)，閾值一般設置為0.5。PR曲線則是不斷調(diào)整閾值得到準確率和召回率的曲線，我們考察的是曲線包圍面積，曲線的面積約到則表示這個模型越好。

PR曲線

ROC曲線與AUC

ROC曲線和PR曲線類似，不同的是考察的真陽性率與假陽性率。

真陽性率 = 真陽/(真陽+假陰)

假陽性率 = 假陽/(假陽+真陰)

曲線和PR曲線類似，下方面積被稱為AUC。

ROC曲線

下面的代碼計算了PR和ROC下方的面積，Spark中有類可以很方便地計算這些值。

    val metrics = Seq(lrModel, svmModel).map {model =>
      val scoreAndLabels = data.map {lp =>
        (model.predict(lp.features), lp.label)
      }
      val metrics = new BinaryClassificationMetrics(scoreAndLabels)
      (model.getClass.getSimpleName(), metrics.areaUnderPR(), metrics.areaUnderROC())
    }
    val nbMetrics = Seq(nbModel).map {model =>
      val scoreAndLabels = nbData.map {lp =>
        (model.predict(lp.features), lp.label)
      }
      val metrics = new BinaryClassificationMetrics(scoreAndLabels)
      (model.getClass.getSimpleName(), metrics.areaUnderPR(), metrics.areaUnderROC())
    }
    val dtMetrics = Seq(dtModel).map {model =>
      val scoreAndLabels = data.map {lp =>
        val score = model.predict(lp.features)
        (if (score > 0.5) 1.0 else 0.0, lp.label)
      }
      val metrics = new BinaryClassificationMetrics(scoreAndLabels)
      (model.getClass.getSimpleName(), metrics.areaUnderPR(), metrics.areaUnderROC())
    }
    val allMetrics = metrics ++ nbMetrics ++ dtMetrics
    allMetrics.foreach {case (m, pr, roc) =>
      println(f"$m, Area under PR: ${pr * 100}%2.4f%%, Area under ROC: ${roc * 100}%2.4f%%")}

結(jié)果如下：

LogisticRegressionModel, Area under PR: 75.6759%, Area under ROC: 50.1418%
SVMModel, Area under PR: 75.6759%, Area under ROC: 50.1418%
NaiveBayesModel, Area under PR: 68.0851%, Area under ROC: 58.3559%
DecisionTreeModel, Area under PR: 74.3081%, Area under ROC: 64.8837%

第4步：改進模型性能

我們可以發(fā)現(xiàn)，我們訓練出來的模型性能不好，僅比隨機判別好一丟丟。我們來做一些常識來改進它們。

特征標準化

我們把每一種特征都標準化為均值為0，方差為1。當然Spark為我們提供了函數(shù)。注意，標準化不是指每一條數(shù)據(jù)均值為0，而是指某一種特征被標準化，比如年齡。

    val vectors = data.map(lp => lp.features)
    val scaler = new StandardScaler(withMean = true, withStd = true).fit(vectors)
    val scaledData = data.map(lp => LabeledPoint(lp.label, scaler.transform(lp.features)))

在邏輯回歸模型上做個測試：

    val lrModelScaled = LogisticRegressionWithSGD.train(scaledData, numIterations)
    val lrTotalCorrectScaled = scaledData.map { point =>
      if (lrModelScaled.predict(point.features) == point.label) 1 else 0
    }.sum()
    val lrAccuracyScaled = lrTotalCorrectScaled / numData
    val lrPredictionsVsTrue = scaledData.map { point =>
      (lrModelScaled.predict(point.features), point.label)
    }
    val lrMetricsScaled = new BinaryClassificationMetrics(lrPredictionsVsTrue)
    val lrPr = lrMetricsScaled.areaUnderPR
    val lrRoc = lrMetricsScaled.areaUnderROC
    println("Normalize the training data:")
    println(f"${lrModelScaled.getClass.getSimpleName}\n" +
      f"Accuracy: ${lrAccuracyScaled * 100}%2.4f%%\nArea under PR: " +
      f"${lrPr * 100.0}%2.4f%%\nArea under ROC: ${lrRoc * 100.0}%2.4f%%")

結(jié)果為：

Normalize the training data:
LogisticRegressionModel
Accuracy: 62.0419%
Area under PR: 72.7254%
Area under ROC: 61.9663%

效果提升非常明顯，所以：對邏輯回歸，SVM而言，特征標準化非常重要；而決策樹和樸素貝葉斯則不受影響。

加入類別特征

我們還記得我們在訓練時忽略了訓練數(shù)據(jù)的第四項，代表了頁面的類別。我們來把它加入訓練數(shù)據(jù)。還記得方法在系列第三篇文章中有介紹，先統(tǒng)計一共有多少不同類別，再把它映射成one hot的特征向量。

我們加入類別特征，并在邏輯回歸模型上作測試：

    val categories = records.map(r => r(3)).distinct.collect.zipWithIndex.toMap
    val numCategories = categories.size
    val dataCategories = records.map { r =>
      val trimmed = r.map(_.replaceAll("\"", ""))
      val label = trimmed(r.size - 1).toInt
      val categoryIdx = categories(r(3))
      val categoryFeatures = Array.ofDim[Double](numCategories)
      categoryFeatures(categoryIdx) = 1.0
      val otherFeatures = trimmed.slice(4, r.size - 1).map(d => if (d == "?") 0.0 else d.toDouble)
      val features = categoryFeatures ++ otherFeatures
      LabeledPoint(label, Vectors.dense(features))
    }
    val scalerCats = new StandardScaler(withMean = true, withStd = true).fit(dataCategories.map(lp => lp.features))
    val scaledDataCats = dataCategories.map(lp => LabeledPoint(lp.label, scalerCats.transform(lp.features)))
    val lrModelScaledCats = LogisticRegressionWithSGD.train(scaledDataCats, numIterations)
    val lrTotalCorrectScaledCats = scaledDataCats.map { point =>
      if (lrModelScaledCats.predict(point.features) == point.label) 1 else 0
    }.sum
    val lrAccuracyScaledCats = lrTotalCorrectScaledCats / numData
    val lrPredictionsVsTrueCats = scaledDataCats.map { point =>
      (lrModelScaledCats.predict(point.features), point.label)
    }
    val lrMetricsScaledCats = new BinaryClassificationMetrics(lrPredictionsVsTrueCats)
    val lrPrCats = lrMetricsScaledCats.areaUnderPR
    val lrRocCats = lrMetricsScaledCats.areaUnderROC
    println("Add category feature:")
    println(f"${lrModelScaledCats.getClass.getSimpleName}\nAccuracy: " +
      f"${lrAccuracyScaledCats * 100}%2.4f%%\nArea under PR: " +
      f"${lrPrCats * 100.0}%2.4f%%\nArea under ROC: ${lrRocCats * 100.0}%2.4f%%")

結(jié)果為：

Add category feature:
LogisticRegressionModel
Accuracy: 66.5720%
Area under PR: 75.7964%
Area under ROC: 66.5483%

性能進一步得到提升。

第5步：模型參數(shù)調(diào)優(yōu)

之前我們說過，模型的參數(shù)我們都選了默認。實際上，好的參數(shù)當然會使效果變好。參數(shù)調(diào)優(yōu)必須使用交叉驗證。于是我們把訓練集分成60%的訓練集和40%的測試集。

    val trainTestSplit = scaledDataCats.randomSplit(Array(0.6, 0.4), seed = 123)
    val train = trainTestSplit(0)
    val test = trainTestSplit(1)

之后我們?yōu)檫壿嫽貧w加入，L2正則化，即損失函數(shù)要加上所有參數(shù)的平方。并調(diào)整L2正則化的比重。代碼如下，我們首先構(gòu)造了兩個函數(shù)來方便地構(gòu)造與測試模型：

    def trainWithParams(input: RDD[LabeledPoint], regParam: Double,
                            numIterations: Int, updater: Updater, stepSize: Double) = {
      val lr = new LogisticRegressionWithSGD()
      lr.optimizer.setRegParam(regParam).setUpdater(updater).setStepSize(stepSize)
      lr.run(input)
    }
    def createMetrics(label: String, data: RDD[LabeledPoint], model: ClassificationModel) = {
      val scoreAndLabels = data.map {point =>
        (model.predict(point.features), point.label)
      }
      val metrics = new BinaryClassificationMetrics(scoreAndLabels)
      (label, metrics.areaUnderROC)
    }
    scaledDataCats.cache
    val trainTestSplit = scaledDataCats.randomSplit(Array(0.6, 0.4), seed = 123)
    val train = trainTestSplit(0)
    val test = trainTestSplit(1)
    val regResultsTest = Seq(0.0, 0.001, 0.0025, 0.005, 0.01).map {param =>
      val model = trainWithParams(train, param, numIterations, new SquaredL2Updater, 1.0)
      createMetrics(s"$param L2 regularization parameter", train, model)
    }
    regResultsTest.foreach { case (param, auc) => println(f"$param, AUC = ${auc * 100}%2.6f%%") }

我們僅僅考察了AUC，結(jié)果為：

0.0 L2 regularization parameter, AUC = 66.083019%
0.001 L2 regularization parameter, AUC = 66.128304%
0.0025 L2 regularization parameter, AUC = 66.106659%
0.005 L2 regularization parameter, AUC = 66.108655%
0.01 L2 regularization parameter, AUC = 66.181573%

可見，加入L2正則對模型的效果還是有提升的。

理論上，所有涉及到的參數(shù)比如訓練步長，optimizer都要交叉驗證進行調(diào)參。