from pyspark.ml.linalg import Vectors
from pyspark.ml.classification import LogisticRegression
# 从 (标签, 特征) 元组列表准备训练数据。
training = spark.createDataFrame([
(1.0, Vectors.dense([0.0, 1.1, 0.1])),
(0.0, Vectors.dense([2.0, 1.0, -1.0])),
(0.0, Vectors.dense([2.0, 1.3, 1.0])),
(1.0, Vectors.dense([0.0, 1.2, -0.5]))], ["label", "features"])
# 创建一个 LogisticRegression 实例。此实例是一个估计器。
lr = LogisticRegression(maxIter=10, regParam=0.01)
# 打印参数、文档和任何默认值。
print("LogisticRegression 参数:\n" + lr.explainParams() + "\n")
# 学习 LogisticRegression 模型。这使用存储在 lr 中的参数。
model1 = lr.fit(training)
# 因为 model1 是一个模型（即由估计器生成的变换器），
# 我们可以查看它在 fit() 期间使用的参数。
# 这将打印参数（名称: 值）对，其中名称是此
# LogisticRegression 实例的唯一 ID。
print("模型 1 使用的参数: ")
print(model1.extractParamMap())
# 我们也可以使用 Python 字典指定参数作为 paramMap
paramMap = {lr.maxIter: 20}
paramMap[lr.maxIter] = 30 # 指定 1 个参数，覆盖原来的 maxIter。
# 指定多个参数。
paramMap.update({lr.regParam: 0.1, lr.threshold: 0.55}) # type: ignore

# 您可以组合 paramMaps，它们是 Python 字典。
# 更改输出列名称
paramMap2 = {lr.probabilityCol: "myProbability"}
paramMapCombined = paramMap.copy()
paramMapCombined.update(paramMap2) # type: ignore

# 现在使用 paramMapCombined 参数学习新模型。
# paramMapCombined 覆盖之前通过 lr.set* 方法设置的所有参数。
model2 = lr.fit(training, paramMapCombined)
print("模型 2 使用的参数: ")
print(model2.extractParamMap())
# 准备测试数据
test = spark.createDataFrame([
(1.0, Vectors.dense([-1.0, 1.5, 1.3])),
(0.0, Vectors.dense([3.0, 2.0, -0.1])),
(1.0, Vectors.dense([0.0, 2.2, -1.5]))], ["label", "features"])
# 使用 Transformer.transform() 方法对测试数据进行预测。
# LogisticRegression.transform 将仅使用 'features' 列。
# 注意 model2.transform() 输出一个 "myProbability" 列，而不是通常的
# 'probability' 列，因为我们之前重命名了 lr.probabilityCol 参数。
prediction = model2.transform(test)
result = prediction.select("features", "label", "myProbability", "prediction") \
    .collect()
for row in result:
print("features=%s, label=%s -> prob=%s, prediction=%s"
% (row.features, row.label, row.myProbability, row.prediction))

import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.linalg.{Vector, Vectors}
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.sql.Row
// 从(label, features)元组的列表中准备训练数据。
val training = spark.createDataFrame(Seq(
(1.0, Vectors.dense(0.0, 1.1, 0.1)),
(0.0, Vectors.dense(2.0, 1.0, -1.0)),
(0.0, Vectors.dense(2.0, 1.3, 1.0)),
(1.0, Vectors.dense(0.0, 1.2, -0.5))
)).toDF("label", "features")
// 创建一个LogisticRegression实例。这个实例是一个估计器。
val lr = new LogisticRegression()
// 输出参数、文档及任何默认值。
println(s"LogisticRegression参数:\n ${lr.explainParams()}\n")
// 我们可以使用设置方法来设置参数。
lr.setMaxIter(10)
.setRegParam(0.01)
// 学习一个LogisticRegression模型。 这使用存储在lr中的参数。
val model1 = lr.fit(training)
// 由于model1是一个模型（即由估计器生成的转换器），
// 我们可以查看它在fit()期间使用的参数。
// 这将打印参数（名称：值）对，名称为此
// LogisticRegression实例的唯一ID。
println(s"模型1是采用以下参数训练的: ${model1.parent.extractParamMap}")
// 我们也可以使用ParamMap指定参数，
// ParamMap支持多种指定参数的方法。
val paramMap = ParamMap(lr.maxIter -> 20)
.put(lr.maxIter, 30) // 指定1个参数。这将覆盖原来的maxIter。
.put(lr.regParam -> 0.1, lr.threshold -> 0.55) // 指定多个参数。
// 也可以组合多个ParamMaps。
val paramMap2 = ParamMap(lr.probabilityCol -> "myProbability") // 更改输出列名。
val paramMapCombined = paramMap ++ paramMap2
// 现在使用paramMapCombined参数学习新的模型。
// paramMapCombined会覆盖之前通过lr.set*方法设置的所有参数。
val model2 = lr.fit(training, paramMapCombined)
println(s"模型2是采用以下参数训练的: ${model2.parent.extractParamMap}")
// 准备测试数据。
val test = spark.createDataFrame(Seq(
(1.0, Vectors.dense(-1.0, 1.5, 1.3)),
(0.0, Vectors.dense(3.0, 2.0, -0.1)),
(1.0, Vectors.dense(0.0, 2.2, -1.5))
)).toDF("label", "features")
// 使用Transformer.transform()方法对测试数据进行预测。
// LogisticRegression.transform将仅使用"features"列。
// 请注意model2.transform()输出的是'myProbability'列，而不是通常的
// 'probability'列，因为我们之前重命名了lr.probabilityCol参数。
model2.transform(test)
.select("features", "label", "myProbability", "prediction")
.collect()
.foreach { case Row(features: Vector, label: Double, prob: Vector, prediction: Double) =>
println(s"($features, $label) -> prob=$prob, prediction=$prediction")
}

import java.util.Arrays;
import java.util.List;
import org.apache.spark.ml.classification.LogisticRegression;
import org.apache.spark.ml.classification.LogisticRegressionModel;
import org.apache.spark.ml.linalg.VectorUDT;
import org.apache.spark.ml.linalg.Vectors;
import org.apache.spark.ml.param.ParamMap;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
// 准备训练数据。
List<Row> dataTraining = Arrays.asList(
RowFactory.create(1.0, Vectors.dense(0.0, 1.1, 0.1)),
RowFactory.create(0.0, Vectors.dense(2.0, 1.0, -1.0)),
RowFactory.create(0.0, Vectors.dense(2.0, 1.3, 1.0)),
RowFactory.create(1.0, Vectors.dense(0.0, 1.2, -0.5))
);
StructType schema = new StructType(new StructField[]{
new StructField("label", DataTypes.DoubleType, false, Metadata.empty()),
new StructField("features", new VectorUDT(), false, Metadata.empty())
});
Dataset<Row> training = spark.createDataFrame(dataTraining, schema);
// 创建一个 LogisticRegression 实例。这个实例是一个估计器。
LogisticRegression lr = new LogisticRegression();
// 打印参数，文档和任何默认值。
System.out.println("LogisticRegression 参数:\n" + lr.explainParams() + "\n");
// 我们可以使用设置方法设置参数。
lr.setMaxIter(10).setRegParam(0.01);
// 学习一个 LogisticRegression 模型。 这使用储存在 lr 中的参数。
LogisticRegressionModel model1 = lr.fit(training);
// 由于 model1 是一个模型（即由估计器生成的转换器），
// 我们可以查看它在 fit() 期间使用的参数。
// 这将打印参数（名称：值）对，其中名称是此
// LogisticRegression 实例的唯一 ID。
System.out.println("Model 1 是使用参数拟合的: " + model1.parent().extractParamMap());
// 我们也可以通过 ParamMap 来指定参数。
ParamMap paramMap = new ParamMap()
.put(lr.maxIter().w(20)) // 指定 1 个参数。
.put(lr.maxIter(), 30) // 这将覆盖原始的 maxIter。
.put(lr.regParam().w(0.1), lr.threshold().w(0.55)); // 指定多个参数。
// 还可以组合 ParamMaps。
ParamMap paramMap2 = new ParamMap()
.put(lr.probabilityCol().w("myProbability")); // 更改输出列名称
ParamMap paramMapCombined = paramMap.$plus$plus(paramMap2);
// 现在使用 paramMapCombined 参数学习一个新模型。
// paramMapCombined 覆盖了之前通过 lr.set* 方法设置的所有参数。
LogisticRegressionModel model2 = lr.fit(training, paramMapCombined);
System.out.println("Model 2 是使用参数拟合的: " + model2.parent().extractParamMap());
// 准备测试文档。
List<Row> dataTest = Arrays.asList(
RowFactory.create(1.0, Vectors.dense(-1.0, 1.5, 1.3)),
RowFactory.create(0.0, Vectors.dense(3.0, 2.0, -0.1)),
RowFactory.create(1.0, Vectors.dense(0.0, 2.2, -1.5))
);
Dataset<Row> test = spark.createDataFrame(dataTest, schema);
// 使用 Transformer.transform() 方法对测试文档进行预测。
// LogisticRegression.transform 将只使用 'features' 列。
// 请注意 model2.transform() 输出 'myProbability' 列，而不是通常的
// 'probability' 列，因为我们之前重命名了 lr.probabilityCol 参数。
Dataset<Row> results = model2.transform(test);
Dataset<Row> rows = results.select("features", "label", "myProbability", "prediction");
for (Row r: rows.collectAsList()) {
System.out.println("(" + r.get(0) + ", " + r.get(1) + ") -> prob=" + r.get(2)
+ ", prediction=" + r.get(3));
}

from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import HashingTF, Tokenizer
# 从 (id, text, label) 元组列表准备训练文档。
training = spark.createDataFrame([
(0, "a b c d e spark", 1.0),
(1, "b d", 0.0),
(2, "spark f g h", 1.0),
(3, "hadoop mapreduce", 0.0)
], ["id", "text", "label"])
# 配置一个机器学习管道，它由三个阶段组成：分词器、HashingTF 和逻辑回归。
tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol="features")
lr = LogisticRegression(maxIter=10, regParam=0.001)
pipeline = Pipeline(stages=[tokenizer, hashingTF, lr])
# 将管道拟合到训练文档。
model = pipeline.fit(training)
# 准备测试文档，这些文档没有标签 (id, text) 元组。
test = spark.createDataFrame([
(4, "spark i j k"),
(5, "l m n"),
(6, "spark hadoop spark"),
(7, "apache hadoop")
], ["id", "text"])
# 对测试文档进行预测并打印感兴趣的列。
prediction = model.transform(test)
selected = prediction.select("id", "text", "probability", "prediction")
for row in selected.collect():
rid, text, prob, prediction = row
print(
"(%d, %s) --> prob=%s, prediction=%f" % (
rid, text, str(prob), prediction # type: ignore
 )
)

import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.feature.{HashingTF, Tokenizer}
import org.apache.spark.ml.linalg.Vector
import org.apache.spark.sql.Row
// 从一个 (id, text, label) 元组的列表中准备训练文档。
val training = spark.createDataFrame(Seq(
(0L, "a b c d e spark", 1.0),
(1L, "b d", 0.0),
(2L, "spark f g h", 1.0),
(3L, "hadoop mapreduce", 0.0)
)).toDF("id", "text", "label")
// 配置一个 ML 管道，它由三个阶段组成：分词器、hashingTF 和 lr。
val tokenizer = new Tokenizer()
.setInputCol("text")
.setOutputCol("words")
val hashingTF = new HashingTF()
.setNumFeatures(1000)
.setInputCol(tokenizer.getOutputCol)
.setOutputCol("features")
val lr = new LogisticRegression()
.setMaxIter(10)
.setRegParam(0.001)
val pipeline = new Pipeline()
.setStages(Array(tokenizer, hashingTF, lr))
// 将管道拟合到训练文档上。
val model = pipeline.fit(training)
// 现在我们可以选择将拟合的管道保存到磁盘
model.write.overwrite().save("/tmp/spark-logistic-regression-model")
// 我们也可以将这个未拟合的管道保存到磁盘
pipeline.write.overwrite().save("/tmp/unfit-lr-model")
// 并在生产中重新加载它
val sameModel = PipelineModel.load("/tmp/spark-logistic-regression-model")
// 准备测试文档，未标记的 (id, text) 元组。
val test = spark.createDataFrame(Seq(
(4L, "spark i j k"),
(5L, "l m n"),
(6L, "spark hadoop spark"),
(7L, "apache hadoop")
)).toDF("id", "text")
// 对测试文档进行预测。
model.transform(test)
.select("id", "text", "probability", "prediction")
.collect()
.foreach { case Row(id: Long, text: String, prob: Vector, prediction: Double) =>
println(s"($id, $text) --> prob=$prob, prediction=$prediction")
}

import java.util.Arrays;
import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.LogisticRegression;
import org.apache.spark.ml.feature.HashingTF;
import org.apache.spark.ml.feature.Tokenizer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
// 准备训练文档，这些文档带有标签。
Dataset<Row> training = spark.createDataFrame(Arrays.asList(
new JavaLabeledDocument(0L, "a b c d e spark", 1.0),
new JavaLabeledDocument(1L, "b d", 0.0),
new JavaLabeledDocument(2L, "spark f g h", 1.0),
new JavaLabeledDocument(3L, "hadoop mapreduce", 0.0)
), JavaLabeledDocument.class);
// 配置一个机器学习管道，该管道由三个阶段组成：分词器、HashingTF 和 lr。
Tokenizer tokenizer = new Tokenizer()
.setInputCol("text")
.setOutputCol("words");
HashingTF hashingTF = new HashingTF()
.setNumFeatures(1000)
.setInputCol(tokenizer.getOutputCol())
.setOutputCol("features");
LogisticRegression lr = new LogisticRegression()
.setMaxIter(10)
.setRegParam(0.001);
Pipeline pipeline = new Pipeline()
.setStages(new PipelineStage[] {tokenizer, hashingTF, lr});
// 将管道拟合到训练文档上。
PipelineModel model = pipeline.fit(training);
// 准备测试文档，这些文档没有标签。
Dataset<Row> test = spark.createDataFrame(Arrays.asList(
new JavaDocument(4L, "spark i j k"),
new JavaDocument(5L, "l m n"),
new JavaDocument(6L, "spark hadoop spark"),
new JavaDocument(7L, "apache hadoop")
), JavaDocument.class);
// 对测试文档进行预测。
Dataset<Row> predictions = model.transform(test);
for (Row r : predictions.select("id", "text", "probability", "prediction").collectAsList()) {
System.out.println("(" + r.get(0) + ", " + r.get(1) + ") --> prob=" + r.get(2)
+ ", prediction=" + r.get(3));
}