Spark ML 构建分类模型

一、数据的加载 1.1、数据的字段

"url"   "urlid" "boilerplate"   "alchemy_category"      "alchemy_category_score"        "avglinksize"   "commonlinkratio_1"     "commonlinkratio_2"     "commonlinkratio_3" "commonlinkratio_4"     "compression_ratio"     "embed_ratio"   "framebased"    "frameTagRatio" "hasDomainLink" "html_ratio"    "image_ratio"       "is_news"       "lengthyLinkDomain"     "linkwordscore" "news_front_page"       "non_markup_alphanum_characters"        "numberOfLinks" "numwords_in_url"   "parametrizedLinkRatio" "spelling_errors_ratio" "label"

下面是部分字段说明

1.2、数据的加载

        val data = spark.sparkContext.textFile("C:\\Users\\12285\\Downloads\\train.tsv")
                .filter(!_.contains("alchemy_category_score"))
                .map(line => line.split("\t"))

二、数据预处理 2.1、数据清洗

        val dataVecs = data.map(t => {
            // 去掉引号
            val fields = t.map(_.replaceAll("\"", ""))
            // 将最后一列(标签列)转为整数
            val label = fields(fields.size - 1).toInt
            // 把第四列中的"?"转为0.0
            val features = fields.slice(4, fields.size - 1).map(d => if (d == "?") 0.0 else d.toDouble)
            // 打标签， 将标签及特征转换为LabeledPoint
            LabeledPoint(label, Vectors.dense(features))
        })

2.2、考虑特殊情况， 贝叶斯算法中数据不小于0，所以需要做些处理

        // 朴素贝叶斯数据集
        val nbDataVecs = data.map(t => {
            // 去掉引号
            val fields = t.map(_.replaceAll("\"", ""))
            // 将最后一列(标签列)转为整数
            val label = fields(fields.size - 1).toInt
            val features = fields.slice(4, fields.size - 1)
                    .map(d => if (d == "?") 0.0 else d.toDouble)
                    .map(d => if (d < 0.0) 0.0 else d) // 朴素贝叶斯算法时，数据需不小于0

            LabeledPoint(label, Vectors.dense(features))
        })

2.3、创建DataFrame

        // 创建DataFrame
        val df = spark.createDataFrame(dataVecs)
        val nbDf = spark.createDataFrame(nbDataVecs)

2.4、划分训练集， 测试集

        // 训练集， 测试集
        val Array(trainingDf, testDf) = df.randomSplit(Array(0.8, 0.2))
        val Array(nbTrainingDf, nbTestDf) = nbDf.randomSplit(Array(0.8, 0.2))

		// 由于后续使用网格参数和交叉验证的时候，需要多次使用到训练集和测试集，所以将这两者载入内存，可大大提高性能。
        trainingDf.persist()
        testDf.persist()
        nbTrainingDf.persist()
        nbTestDf.persist()

三、模型 3.1、贝叶斯模型 3.1.1、创建贝叶斯模型

        // Estimator
        val nb = new NaiveBayes().setLabelCol("label").setFeaturesCol("features")

3.1.2、训练模型

            // 训练数据
            val nbModel = nb.fit(nbTrainingDf)

3.1.2、使用测试数据预测

            // 预测数据
            val nbPrediction = nbModel.transform(nbTestDf)

3.1.3、统计朴素贝叶斯准确性

            //t1 存放预测值的数组,t2存放测试数据标签值 t3存放测试数据总行数
            val (t1, t2, t3) = (nbPrediction.select("prediction").collect(),
                                nbTestDf.select("label").collect(),
                                nbTestDf.count().toInt)
            var t4 = 0
            for (i  if (d == "?") 0.0 else d.toDouble)
                .map(d => if (d < 0.0) 0.0 else d) // 朴素贝叶斯算法时，数据需不小于0

            LabeledPoint(label, Vectors.dense(features))
        })


        // 创建DataFrame
        val df = spark.createDataFrame(dataVecs)
        val nbDf = spark.createDataFrame(nbDataVecs)
        df.show(10)

        // 训练集， 测试集
        val Array(trainingDf, testDf) = df.randomSplit(Array(0.8, 0.2))
        val Array(nbTrainingDf, nbTestDf) = nbDf.randomSplit(Array(0.8, 0.2))
        // 由于后续使用网格参数和交叉验证的时候，需要多次使用到训练集和测试集，所以将这两者载入内存，可大大提高性能。
        trainingDf.persist()
        testDf.persist()
        nbTrainingDf.persist()
        nbTestDf.persist()


//        testBayes()
        testLr()
//        testDt()


        def testBayes() = {
            // Estimator
            val nb = new NaiveBayes().setLabelCol("label").setFeaturesCol("features")
            // 训练数据
            val nbModel = nb.fit(nbTrainingDf)
            // 预测数据
            val nbPrediction = nbModel.transform(nbTestDf)
            nbPrediction.show(10)

            //t1 存放预测值的数组,t2存放测试数据标签值 t3存放测试数据总行数
            val (t1, t2, t3) = (nbPrediction.select("prediction").collect(),
                                nbTestDf.select("label").collect(),
                                nbTestDf.count().toInt)
            var t4 = 0
            for (i