# 梯度上升算法-计算回归系数

# 每个回归系数初始化为1

# 重复R次：

#    计算整个数据集的梯度

#    使用α*梯度更新回归系数的向量

#    返回回归系数

def gradAscent(dataMatIn, classLabels,alpha=0.001,maxCycles = ):

    dataMatrix = mat(dataMatIn)     #转换为numpy数据类型

    labelMat = mat(classLabels).transpose()

    m,n = shape(dataMatrix)

    maxCycles =

    weights = ones((n,))

    for k in range(maxCycles):

        h = sigmoid(dataMatrix*weights)

        error = (labelMat - h)

        #计算真实类别与预测类别的差值，按照该差值的方向调整回归系数

        weights = weights + alpha* dataMatrix.transpose() * error

    return weights  

LogisticRegression lr=new LogisticRegression()

                .setMaxIter()

                .setRegParam(0.3)

                .setElasticNetParam(0.2)

                .setThreshold(0.5);

//获得回归模型训练的Summary

LogisticRegressionTrainingSummary trainingSummary = lrModel.summary();

// Obtain the loss per iteration.

//每次迭代的损失，一般会逐渐减小

double[] objectiveHistory = trainingSummary.objectiveHistory();

for (double lossPerIteration : objectiveHistory) {

  System.out.println(lossPerIteration);

}

// Obtain the metrics useful to judge performance on test data.

// We cast the summary to a BinaryLogisticRegressionSummary since the problem is a binary

// classification problem.

//强制类型转换为二类LR的Summary，然后就可以用混淆矩阵,ROC等评估方法了。Spark2.0还无法针对多类

BinaryLogisticRegressionSummary binarySummary =

  (BinaryLogisticRegressionSummary) trainingSummary;

// Obtain the receiver-operating characteristic as a dataframe and areaUnderROC.

Dataset<Row> roc = binarySummary.roc();//获得ROC

roc.show();//显示ROC数据表，可以用这个数据自己画ROC曲线

roc.select("FPR").show();

System.out.println(binarySummary.areaUnderROC());//AUC

// Get the threshold corresponding to the maximum F-Measure and rerun LogisticRegression with

// this selected threshold.

//不同的阈值，计算不同的F1，然后通过最大的F1找出并重设模型的最佳阈值。

Dataset<Row> fMeasure = binarySummary.fMeasureByThreshold();

double maxFMeasure = fMeasure.select(functions.max("F-Measure")).head().getDouble();//获得最大的F1值

double bestThreshold = fMeasure.where(fMeasure.col("F-Measure").equalTo(maxFMeasure))

  .select("threshold").head().getDouble();//找出最大F1值对应的阈值（最佳阈值）

lrModel.setThreshold(bestThreshold);//并将模型的Threshold设置为选择出来的最佳分类阈值

package my.spark.ml.practice.classification;

import org.apache.log4j.Level;

import org.apache.log4j.Logger;

import org.apache.spark.ml.classification.BinaryLogisticRegressionTrainingSummary;

import org.apache.spark.ml.classification.LogisticRegression;

import org.apache.spark.ml.classification.LogisticRegressionModel;

import org.apache.spark.ml.classification.LogisticRegressionTrainingSummary;

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SparkSession;

import org.apache.spark.sql.functions;

public class myLogisticRegression {

    public static void main(String[] args) {

        SparkSession spark=SparkSession

                .builder()

                .appName("LR")

                .master("local[4]")

                .config("spark.sql.warehouse.dir","file///:G:/Projects/Java/Spark/spark-warehouse" )

                .getOrCreate();

        String path="G:/Projects/CgyWin64/home/pengjy3/softwate/spark-2.0.0-bin-hadoop2.6/"

                + "data/mllib/sample_libsvm_data.txt";

        //屏蔽日志

        Logger.getLogger("org.apache.spark").setLevel(Level.WARN);

        Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF);       

        //Load trainning data

        Dataset<Row> trainning_dataFrame=spark.read().format("libsvm").load(path);      

        LogisticRegression lr=new LogisticRegression()

                .setMaxIter()

                .setRegParam(0.3)

                .setElasticNetParam(0.2)

                .setThreshold(0.5);

        //fit the model

        LogisticRegressionModel lrModel=lr.fit(trainning_dataFrame);

        //print the coefficients and intercept for logistic regression

        System.out.println

        ("Coefficient:"+lrModel.coefficients()+"Itercept"+lrModel.intercept());

        //Extract the summary from the returned LogisticRegressionModel

        LogisticRegressionTrainingSummary summary=lrModel.summary();

        //Obtain the loss per iteration.

        double[] objectiveHistory=summary.objectiveHistory();

        for(double lossPerIteration:objectiveHistory){

            System.out.println(lossPerIteration);

        }

        // Obtain the metrics useful to judge performance on test data.

        // We cast the summary to a BinaryLogisticRegressionSummary since the problem is a binary

        // classification problem.

        BinaryLogisticRegressionTrainingSummary binarySummary=

                (BinaryLogisticRegressionTrainingSummary)summary;

        //Obtain the receiver-operating characteristic as a dataframe and areaUnderROC.

        Dataset<Row> roc=binarySummary.roc();

        roc.show((int) roc.count());//显示全部的信息，roc.show()默认只显示20行

        roc.select("FPR").show();

        System.out.println(binarySummary.areaUnderROC());

        // Get the threshold corresponding to the maximum F-Measure and rerun LogisticRegression with

        // this selected threshold.

        Dataset<Row> fMeasure = binarySummary.fMeasureByThreshold();

        double maxFMeasure = fMeasure.select(functions.max("F-Measure")).head().getDouble();

        double bestThreshold = fMeasure.where(fMeasure.col("F-Measure").equalTo(maxFMeasure))

          .select("threshold").head().getDouble();

        lrModel.setThreshold(bestThreshold);

    }

}

//自定义计算accuracy，

        Dataset<Row> predictDF=naiveBayesModel.transform(test);    

        double total=(double) predictDF.count();

        Encoder<Double> doubleEncoder=Encoders.DOUBLE();

        Dataset<Double> accuracyDF=predictDF.map(new MapFunction<Row,Double>() {

            @Override

            public Double call(Row row) throws Exception {

                if((double)row.get()==(double)row.get()){return 1.0;}

                else {return 0.0;}

            }

        }, doubleEncoder);

        accuracyDF.createOrReplaceTempView("view");

        double correct=(double) spark.sql("SELECT value FROM view WHERE value=1.0").count();

        System.out.println("accuracy "+(correct/total));