분류와 회귀

분류와 회귀 (Classification and Regression)

이 페이지는 분류(Classification)와 회귀(Regression) 알고리즘을 다뤄요. 또한 선형 메서드, 트리, 앙상블 같은 특정 알고리즘 클래스를 설명하는 섹션도 포함해요.

출처: Classification and Regression

본문

분류(Classification)

로지스틱 회귀(Logistic regression)

로지스틱 회귀는 범주형 응답을 예측하는 인기 있는 방법이에요. 결과의 확률을 예측하는 일반화 선형 모델(Generalized Linear models)의 특수한 경우예요. spark.ml에서 로지스틱 회귀는 이항 로지스틱 회귀(binomial logistic regression)를 사용해 이진 결과를 예측하거나, 다항 로지스틱 회귀(multinomial logistic regression)를 사용해 다중 클래스 결과를 예측하는 데 쓸 수 있어요. family 파라미터로 이 두 알고리즘 중 하나를 선택하거나, 설정하지 않고 Spark이 올바른 변형을 추론하게 할 수 있어요.

family 파라미터를 "multinomial"로 설정하면 다항 로지스틱 회귀를 이진 분류에 사용할 수 있어요. 이 경우 두 세트의 계수와 두 개의 절편이 생성돼요.

상수가 아닌 0이 아닌 컬럼이 있는 데이터셋에서 절편 없이 LogisticRegressionModel을 피팅할 때, Spark MLlib는 상수 0이 아닌 컬럼에 대해 0 계수를 출력해요. 이 동작은 R glmnet과 같지만 LIBSVM과는 달라요.

이항 로지스틱 회귀(Binomial logistic regression)

이항 로지스틱 회귀 구현에 대한 더 많은 배경과 세부 사항은 spark.mllib의 로지스틱 회귀 문서를 참고하세요.

예제

다음 예제는 elastic net 정규화로 이진 분류를 위한 이항·다항 로지스틱 회귀 모델을 훈련하는 방법을 보여줘요. elasticNetParam은 $\alpha$에 해당하고 regParam은 $\lambda$에 해당해요.

파라미터에 대한 자세한 내용은 Python API 문서에서 찾을 수 있어요.

from pyspark.ml.classification import LogisticRegression

# Load training data
training = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

lr = LogisticRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8)

# Fit the model
lrModel = lr.fit(training)

# Print the coefficients and intercept for logistic regression
print("Coefficients: " + str(lrModel.coefficients))
print("Intercept: " + str(lrModel.intercept))

# We can also use the multinomial family for binary classification
mlr = LogisticRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8, family="multinomial")

# Fit the model
mlrModel = mlr.fit(training)

# Print the coefficients and intercepts for logistic regression with multinomial family
print("Multinomial coefficients: " + str(mlrModel.coefficientMatrix))
print("Multinomial intercepts: " + str(mlrModel.interceptVector))

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/logistic_regression_with_elastic_net.py"에서 찾을 수 있어요.

파라미터에 대한 자세한 내용은 Scala API 문서에서 찾을 수 있어요.

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

// Load training data
val training = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

val lr = new LogisticRegression()
  .setMaxIter(10)
  .setRegParam(0.3)
  .setElasticNetParam(0.8)

// Fit the model
val lrModel = lr.fit(training)

// Print the coefficients and intercept for logistic regression
println(s"Coefficients: ${lrModel.coefficients} Intercept: ${lrModel.intercept}")

// We can also use the multinomial family for binary classification
val mlr = new LogisticRegression()
  .setMaxIter(10)
  .setRegParam(0.3)
  .setElasticNetParam(0.8)
  .setFamily("multinomial")

val mlrModel = mlr.fit(training)

// Print the coefficients and intercepts for logistic regression with multinomial family
println(s"Multinomial coefficients: ${mlrModel.coefficientMatrix}")
println(s"Multinomial intercepts: ${mlrModel.interceptVector}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/LogisticRegressionWithElasticNetExample.scala"에서 찾을 수 있어요.

파라미터에 대한 자세한 내용은 Java API 문서에서 찾을 수 있어요.

import org.apache.spark.ml.classification.LogisticRegression;
import org.apache.spark.ml.classification.LogisticRegressionModel;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load training data
Dataset<Row> training = spark.read().format("libsvm")
  .load("data/mllib/sample_libsvm_data.txt");

LogisticRegression lr = new LogisticRegression()
  .setMaxIter(10)
  .setRegParam(0.3)
  .setElasticNetParam(0.8);

// Fit the model
LogisticRegressionModel lrModel = lr.fit(training);

// Print the coefficients and intercept for logistic regression
System.out.println("Coefficients: "
  + lrModel.coefficients() + " Intercept: " + lrModel.intercept());

// We can also use the multinomial family for binary classification
LogisticRegression mlr = new LogisticRegression()
        .setMaxIter(10)
        .setRegParam(0.3)
        .setElasticNetParam(0.8)
        .setFamily("multinomial");

// Fit the model
LogisticRegressionModel mlrModel = mlr.fit(training);

// Print the coefficients and intercepts for logistic regression with multinomial family
System.out.println("Multinomial coefficients: " + lrModel.coefficientMatrix()
  + "\nMultinomial intercepts: " + mlrModel.interceptVector());

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaLogisticRegressionWithElasticNetExample.java"에서 찾을 수 있어요.

파라미터에 대한 자세한 내용은 R API 문서에서 찾을 수 있어요.

# Load training data
df <- read.df("data/mllib/sample_libsvm_data.txt", source = "libsvm")
training <- df
test <- df

# Fit an binomial logistic regression model with spark.logit
model <- spark.logit(training, label ~ features, maxIter = 10, regParam = 0.3, elasticNetParam = 0.8)

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/logit.R"에서 찾을 수 있어요.

spark.ml의 로지스틱 회귀 구현은 훈련 세트에 대한 모델 요약을 추출하는 것도 지원해요. LogisticRegressionSummary에 DataFrame으로 저장된 예측과 지표는 @transient로 주석 처리되어 드라이버에서만 사용할 수 있다는 점에 유의하세요.

LogisticRegressionTrainingSummary는 LogisticRegressionModel에 대한 요약을 제공해요. 이진 분류의 경우 ROC 곡선 같은 특정 추가 지표를 사용할 수 있어요. BinaryLogisticRegressionTrainingSummary를 참고하세요.

앞선 예제를 계속해서:

from pyspark.ml.classification import LogisticRegression

# Extract the summary from the returned LogisticRegressionModel instance trained
# in the earlier example
trainingSummary = lrModel.summary

# Obtain the objective per iteration
objectiveHistory = trainingSummary.objectiveHistory
print("objectiveHistory:")
for objective in objectiveHistory:
    print(objective)

# Obtain the receiver-operating characteristic as a dataframe and areaUnderROC.
trainingSummary.roc.show()
print("areaUnderROC: " + str(trainingSummary.areaUnderROC))

# Set the model threshold to maximize F-Measure
fMeasure = trainingSummary.fMeasureByThreshold
maxFMeasure = fMeasure.groupBy().max('F-Measure').select('max(F-Measure)').head()
bestThreshold = fMeasure.where(fMeasure['F-Measure'] == maxFMeasure['max(F-Measure)']) \
    .select('threshold').head()['threshold']
lr.setThreshold(bestThreshold)

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/logistic_regression_summary_example.py"에서 찾을 수 있어요.

LogisticRegressionTrainingSummary는 LogisticRegressionModel에 대한 요약을 제공해요. 이진 분류의 경우 ROC 곡선 같은 특정 추가 지표를 사용할 수 있어요. 이진 요약은 binarySummary 메서드로 접근할 수 있어요. BinaryLogisticRegressionTrainingSummary를 참고하세요.

앞선 예제를 계속해서:

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

// Extract the summary from the returned LogisticRegressionModel instance trained in the earlier
// example
val trainingSummary = lrModel.binarySummary

// Obtain the objective per iteration.
val objectiveHistory = trainingSummary.objectiveHistory
println("objectiveHistory:")
objectiveHistory.foreach(loss => println(loss))

// Obtain the receiver-operating characteristic as a dataframe and areaUnderROC.
val roc = trainingSummary.roc
roc.show()
println(s"areaUnderROC: ${trainingSummary.areaUnderROC}")

// Set the model threshold to maximize F-Measure
val fMeasure = trainingSummary.fMeasureByThreshold
val maxFMeasure = fMeasure.select(max("F-Measure")).head().getDouble(0)
val bestThreshold = fMeasure.where($"F-Measure" === maxFMeasure)
  .select("threshold").head().getDouble(0)
lrModel.setThreshold(bestThreshold)

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/LogisticRegressionSummaryExample.scala"에서 찾을 수 있어요.

LogisticRegressionTrainingSummary는 LogisticRegressionModel에 대한 요약을 제공해요. 이진 분류의 경우 ROC 곡선 같은 특정 추가 지표를 사용할 수 있어요. 이진 요약은 binarySummary 메서드로 접근할 수 있어요. BinaryLogisticRegressionTrainingSummary를 참고하세요.

앞선 예제를 계속해서:

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.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;

// Extract the summary from the returned LogisticRegressionModel instance trained in the earlier
// example
BinaryLogisticRegressionTrainingSummary trainingSummary = lrModel.binarySummary();

// Obtain the loss per iteration.
double[] objectiveHistory = trainingSummary.objectiveHistory();
for (double lossPerIteration : objectiveHistory) {
  System.out.println(lossPerIteration);
}

// Obtain the receiver-operating characteristic as a dataframe and areaUnderROC.
Dataset<Row> roc = trainingSummary.roc();
roc.show();
roc.select("FPR").show();
System.out.println(trainingSummary.areaUnderROC());

// Get the threshold corresponding to the maximum F-Measure and rerun LogisticRegression with
// this selected threshold.
Dataset<Row> fMeasure = trainingSummary.fMeasureByThreshold();
double maxFMeasure = fMeasure.select(functions.max("F-Measure")).head().getDouble(0);
double bestThreshold = fMeasure.where(fMeasure.col("F-Measure").equalTo(maxFMeasure))
  .select("threshold").head().getDouble(0);
lrModel.setThreshold(bestThreshold);

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaLogisticRegressionSummaryExample.java"에서 찾을 수 있어요.

다항 로지스틱 회귀(Multinomial logistic regression)

다중 클래스 분류는 다항 로지스틱(softmax) 회귀를 통해 지원돼요. 다항 로지스틱 회귀에서 알고리즘은 $K$ 세트의 계수 또는 $K \times J$ 차원의 행렬을 생성하는데, 여기서 $K$는 결과 클래스 수, $J$는 피처 수예요. 알고리즘이 절편 항으로 피팅되면 길이 $K$의 절편 벡터를 사용할 수 있어요.

다항 계수는 coefficientMatrix로, 절편은 interceptVector로 사용할 수 있어요.

다항 family로 훈련된 로지스틱 회귀 모델의 coefficients와 intercept 메서드는 지원되지 않아요. 대신 coefficientMatrix와 interceptVector를 사용하세요.

결과 클래스 $k \in {1, 2, …, K}$의 조건부 확률은 softmax 함수로 모델링돼요.

\[ P(Y=k|\mathbf{X}, \boldsymbol{\beta}_k, \beta_{0k}) = \frac{e^{\boldsymbol{\beta}_k \cdot \mathbf{X} + \beta_{0k}}}{\sum_{k'=0}^{K-1} e^{\boldsymbol{\beta}_{k'} \cdot \mathbf{X} + \beta_{0k'}}} \]

과적합을 제어하기 위해 elastic-net 페널티를 사용해, 다항 응답 모델로 가중 음의 로그 가능도를 최소화해요.

\[ \min_{\beta, \beta_0} -\left[\sum_{i=1}^L w_i \cdot \log P(Y = y_i|\mathbf{x}_i)\right] + \lambda \left[\frac{1}{2}\left(1 - \alpha\right)||\boldsymbol{\beta}||_2^2 + \alpha ||\boldsymbol{\beta}||_1\right] \]

자세한 유도는 여기를 참고하세요.

예제

다음 예제는 elastic net 정규화로 다중 클래스 로지스틱 회귀 모델을 훈련하고, 모델 평가용 다중 클래스 훈련 요약을 추출하는 방법을 보여줘요.

from pyspark.ml.classification import LogisticRegression

# Load training data
training = spark \
    .read \
    .format("libsvm") \
    .load("data/mllib/sample_multiclass_classification_data.txt")

lr = LogisticRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8)

# Fit the model
lrModel = lr.fit(training)

# Print the coefficients and intercept for multinomial logistic regression
print("Coefficients: \n" + str(lrModel.coefficientMatrix))
print("Intercept: " + str(lrModel.interceptVector))

trainingSummary = lrModel.summary

# Obtain the objective per iteration
objectiveHistory = trainingSummary.objectiveHistory
print("objectiveHistory:")
for objective in objectiveHistory:
    print(objective)

# for multiclass, we can inspect metrics on a per-label basis
print("False positive rate by label:")
for i, rate in enumerate(trainingSummary.falsePositiveRateByLabel):
    print("label %d: %s" % (i, rate))

print("True positive rate by label:")
for i, rate in enumerate(trainingSummary.truePositiveRateByLabel):
    print("label %d: %s" % (i, rate))

print("Precision by label:")
for i, prec in enumerate(trainingSummary.precisionByLabel):
    print("label %d: %s" % (i, prec))

print("Recall by label:")
for i, rec in enumerate(trainingSummary.recallByLabel):
    print("label %d: %s" % (i, rec))

print("F-measure by label:")
for i, f in enumerate(trainingSummary.fMeasureByLabel()):
    print("label %d: %s" % (i, f))

accuracy = trainingSummary.accuracy
falsePositiveRate = trainingSummary.weightedFalsePositiveRate
truePositiveRate = trainingSummary.weightedTruePositiveRate
fMeasure = trainingSummary.weightedFMeasure()
precision = trainingSummary.weightedPrecision
recall = trainingSummary.weightedRecall
print("Accuracy: %s\nFPR: %s\nTPR: %s\nF-measure: %s\nPrecision: %s\nRecall: %s"
      % (accuracy, falsePositiveRate, truePositiveRate, fMeasure, precision, recall))

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/multiclass_logistic_regression_with_elastic_net.py"에서 찾을 수 있어요.

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

// Load training data
val training = spark
  .read
  .format("libsvm")
  .load("data/mllib/sample_multiclass_classification_data.txt")

val lr = new LogisticRegression()
  .setMaxIter(10)
  .setRegParam(0.3)
  .setElasticNetParam(0.8)

// Fit the model
val lrModel = lr.fit(training)

// Print the coefficients and intercept for multinomial logistic regression
println(s"Coefficients: \n${lrModel.coefficientMatrix}")
println(s"Intercepts: \n${lrModel.interceptVector}")

val trainingSummary = lrModel.summary

// Obtain the objective per iteration
val objectiveHistory = trainingSummary.objectiveHistory
println("objectiveHistory:")
objectiveHistory.foreach(println)

// for multiclass, we can inspect metrics on a per-label basis
println("False positive rate by label:")
trainingSummary.falsePositiveRateByLabel.zipWithIndex.foreach { case (rate, label) =>
  println(s"label $label: $rate")
}

println("True positive rate by label:")
trainingSummary.truePositiveRateByLabel.zipWithIndex.foreach { case (rate, label) =>
  println(s"label $label: $rate")
}

println("Precision by label:")
trainingSummary.precisionByLabel.zipWithIndex.foreach { case (prec, label) =>
  println(s"label $label: $prec")
}

println("Recall by label:")
trainingSummary.recallByLabel.zipWithIndex.foreach { case (rec, label) =>
  println(s"label $label: $rec")
}


println("F-measure by label:")
trainingSummary.fMeasureByLabel.zipWithIndex.foreach { case (f, label) =>
  println(s"label $label: $f")
}

val accuracy = trainingSummary.accuracy
val falsePositiveRate = trainingSummary.weightedFalsePositiveRate
val truePositiveRate = trainingSummary.weightedTruePositiveRate
val fMeasure = trainingSummary.weightedFMeasure
val precision = trainingSummary.weightedPrecision
val recall = trainingSummary.weightedRecall
println(s"Accuracy: $accuracy\nFPR: $falsePositiveRate\nTPR: $truePositiveRate\n" +
  s"F-measure: $fMeasure\nPrecision: $precision\nRecall: $recall")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/MulticlassLogisticRegressionWithElasticNetExample.scala"에서 찾을 수 있어요.

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;

// Load training data
Dataset<Row> training = spark.read().format("libsvm")
        .load("data/mllib/sample_multiclass_classification_data.txt");

LogisticRegression lr = new LogisticRegression()
        .setMaxIter(10)
        .setRegParam(0.3)
        .setElasticNetParam(0.8);

// Fit the model
LogisticRegressionModel lrModel = lr.fit(training);

// Print the coefficients and intercept for multinomial logistic regression
System.out.println("Coefficients: \n"
        + lrModel.coefficientMatrix() + " \nIntercept: " + lrModel.interceptVector());
LogisticRegressionTrainingSummary trainingSummary = lrModel.summary();

// Obtain the loss per iteration.
double[] objectiveHistory = trainingSummary.objectiveHistory();
for (double lossPerIteration : objectiveHistory) {
    System.out.println(lossPerIteration);
}

// for multiclass, we can inspect metrics on a per-label basis
System.out.println("False positive rate by label:");
int i = 0;
double[] fprLabel = trainingSummary.falsePositiveRateByLabel();
for (double fpr : fprLabel) {
    System.out.println("label " + i + ": " + fpr);
    i++;
}

System.out.println("True positive rate by label:");
i = 0;
double[] tprLabel = trainingSummary.truePositiveRateByLabel();
for (double tpr : tprLabel) {
    System.out.println("label " + i + ": " + tpr);
    i++;
}

System.out.println("Precision by label:");
i = 0;
double[] precLabel = trainingSummary.precisionByLabel();
for (double prec : precLabel) {
    System.out.println("label " + i + ": " + prec);
    i++;
}

System.out.println("Recall by label:");
i = 0;
double[] recLabel = trainingSummary.recallByLabel();
for (double rec : recLabel) {
    System.out.println("label " + i + ": " + rec);
    i++;
}

System.out.println("F-measure by label:");
i = 0;
double[] fLabel = trainingSummary.fMeasureByLabel();
for (double f : fLabel) {
    System.out.println("label " + i + ": " + f);
    i++;
}

double accuracy = trainingSummary.accuracy();
double falsePositiveRate = trainingSummary.weightedFalsePositiveRate();
double truePositiveRate = trainingSummary.weightedTruePositiveRate();
double fMeasure = trainingSummary.weightedFMeasure();
double precision = trainingSummary.weightedPrecision();
double recall = trainingSummary.weightedRecall();
System.out.println("Accuracy: " + accuracy);
System.out.println("FPR: " + falsePositiveRate);
System.out.println("TPR: " + truePositiveRate);
System.out.println("F-measure: " + fMeasure);
System.out.println("Precision: " + precision);
System.out.println("Recall: " + recall);

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaMulticlassLogisticRegressionWithElasticNetExample.java"에서 찾을 수 있어요.

파라미터에 대한 자세한 내용은 R API 문서에서 찾을 수 있어요.

# Load training data
df <- read.df("data/mllib/sample_multiclass_classification_data.txt", source = "libsvm")
training <- df
test <- df

# Fit a multinomial logistic regression model with spark.logit
model <- spark.logit(training, label ~ features, maxIter = 10, regParam = 0.3, elasticNetParam = 0.8)

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/logit.R"에서 찾을 수 있어요.

의사결정 트리 분류기(Decision tree classifier)

의사결정 트리는 인기 있는 분류·회귀 메서드 패밀리예요. spark.ml 구현에 대한 더 많은 정보는 의사결정 트리 섹션에서 찾을 수 있어요.

예제

다음 예제는 LibSVM 형식의 데이터셋을 로드하고, 훈련·테스트 집합으로 나누고, 첫 데이터셋에서 훈련한 뒤 분리된 테스트 집합에서 평가해요. 데이터를 준비하기 위해 두 개의 피처 변환기를 사용해요. 이들은 레이블과 범주형 피처의 범주를 인덱싱해 의사결정 트리 알고리즘이 인식할 수 있는 메타데이터를 DataFrame에 추가해요.

파라미터에 대한 자세한 내용은 Python API 문서에서 찾을 수 있어요.

from pyspark.ml import Pipeline
from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.feature import StringIndexer, VectorIndexer
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# Load the data stored in LIBSVM format as a DataFrame.
data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

# Index labels, adding metadata to the label column.
# Fit on whole dataset to include all labels in index.
labelIndexer = StringIndexer(inputCol="label", outputCol="indexedLabel").fit(data)
# Automatically identify categorical features, and index them.
# We specify maxCategories so features with > 4 distinct values are treated as continuous.
featureIndexer =\
    VectorIndexer(inputCol="features", outputCol="indexedFeatures", maxCategories=4).fit(data)

# Split the data into training and test sets (30% held out for testing)
(trainingData, testData) = data.randomSplit([0.7, 0.3])

# Train a DecisionTree model.
dt = DecisionTreeClassifier(labelCol="indexedLabel", featuresCol="indexedFeatures")

# Chain indexers and tree in a Pipeline
pipeline = Pipeline(stages=[labelIndexer, featureIndexer, dt])

# Train model.  This also runs the indexers.
model = pipeline.fit(trainingData)

# Make predictions.
predictions = model.transform(testData)

# Select example rows to display.
predictions.select("prediction", "indexedLabel", "features").show(5)

# Select (prediction, true label) and compute test error
evaluator = MulticlassClassificationEvaluator(
    labelCol="indexedLabel", predictionCol="prediction", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Test Error = %g " % (1.0 - accuracy))

treeModel = model.stages[2]
# summary only
print(treeModel)

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/decision_tree_classification_example.py"에서 찾을 수 있어요.

파라미터에 대한 자세한 내용은 Scala API 문서에서 찾을 수 있어요.

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.DecisionTreeClassificationModel
import org.apache.spark.ml.classification.DecisionTreeClassifier
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}

// Load the data stored in LIBSVM format as a DataFrame.
val data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

// Index labels, adding metadata to the label column.
// Fit on whole dataset to include all labels in index.
val labelIndexer = new StringIndexer()
  .setInputCol("label")
  .setOutputCol("indexedLabel")
  .fit(data)
// Automatically identify categorical features, and index them.
val featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4) // features with > 4 distinct values are treated as continuous.
  .fit(data)

// Split the data into training and test sets (30% held out for testing).
val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))

// Train a DecisionTree model.
val dt = new DecisionTreeClassifier()
  .setLabelCol("indexedLabel")
  .setFeaturesCol("indexedFeatures")

// Convert indexed labels back to original labels.
val labelConverter = new IndexToString()
  .setInputCol("prediction")
  .setOutputCol("predictedLabel")
  .setLabels(labelIndexer.labelsArray(0))

// Chain indexers and tree in a Pipeline.
val pipeline = new Pipeline()
  .setStages(Array(labelIndexer, featureIndexer, dt, labelConverter))

// Train model. This also runs the indexers.
val model = pipeline.fit(trainingData)

// Make predictions.
val predictions = model.transform(testData)

// Select example rows to display.
predictions.select("predictedLabel", "label", "features").show(5)

// Select (prediction, true label) and compute test error.
val evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("indexedLabel")
  .setPredictionCol("prediction")
  .setMetricName("accuracy")
val accuracy = evaluator.evaluate(predictions)
println(s"Test Error = ${(1.0 - accuracy)}")

val treeModel = model.stages(2).asInstanceOf[DecisionTreeClassificationModel]
println(s"Learned classification tree model:\n ${treeModel.toDebugString}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/DecisionTreeClassificationExample.scala"에서 찾을 수 있어요.

파라미터에 대한 자세한 내용은 Java API 문서에서 찾을 수 있어요.

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.DecisionTreeClassifier;
import org.apache.spark.ml.classification.DecisionTreeClassificationModel;
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
import org.apache.spark.ml.feature.*;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load the data stored in LIBSVM format as a DataFrame.
Dataset<Row> data = spark
  .read()
  .format("libsvm")
  .load("data/mllib/sample_libsvm_data.txt");

// Index labels, adding metadata to the label column.
// Fit on whole dataset to include all labels in index.
StringIndexerModel labelIndexer = new StringIndexer()
  .setInputCol("label")
  .setOutputCol("indexedLabel")
  .fit(data);

// Automatically identify categorical features, and index them.
VectorIndexerModel featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4) // features with > 4 distinct values are treated as continuous.
  .fit(data);

// Split the data into training and test sets (30% held out for testing).
Dataset<Row>[] splits = data.randomSplit(new double[]{0.7, 0.3});
Dataset<Row> trainingData = splits[0];
Dataset<Row> testData = splits[1];

// Train a DecisionTree model.
DecisionTreeClassifier dt = new DecisionTreeClassifier()
  .setLabelCol("indexedLabel")
  .setFeaturesCol("indexedFeatures");

// Convert indexed labels back to original labels.
IndexToString labelConverter = new IndexToString()
  .setInputCol("prediction")
  .setOutputCol("predictedLabel")
  .setLabels(labelIndexer.labelsArray()[0]);

// Chain indexers and tree in a Pipeline.
Pipeline pipeline = new Pipeline()
  .setStages(new PipelineStage[]{labelIndexer, featureIndexer, dt, labelConverter});

// Train model. This also runs the indexers.
PipelineModel model = pipeline.fit(trainingData);

// Make predictions.
Dataset<Row> predictions = model.transform(testData);

// Select example rows to display.
predictions.select("predictedLabel", "label", "features").show(5);

// Select (prediction, true label) and compute test error.
MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("indexedLabel")
  .setPredictionCol("prediction")
  .setMetricName("accuracy");
double accuracy = evaluator.evaluate(predictions);
System.out.println("Test Error = " + (1.0 - accuracy));

DecisionTreeClassificationModel treeModel =
  (DecisionTreeClassificationModel) (model.stages()[2]);
System.out.println("Learned classification tree model:\n" + treeModel.toDebugString());

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaDecisionTreeClassificationExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

# Load training data
df <- read.df("data/mllib/sample_libsvm_data.txt", source = "libsvm")
training <- df
test <- df

# Fit a DecisionTree classification model with spark.decisionTree
model <- spark.decisionTree(training, label ~ features, "classification")

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/decisionTree.R"에서 찾을 수 있어요.

랜덤 포레스트 분류기(Random forest classifier)

랜덤 포레스트는 인기 있는 분류·회귀 메서드 패밀리예요. spark.ml 구현에 대한 더 많은 정보는 랜덤 포레스트 섹션에서 찾을 수 있어요.

예제

다음 예제는 LibSVM 형식의 데이터셋을 로드하고, 훈련·테스트 집합으로 나누고, 첫 데이터셋에서 훈련한 뒤 분리된 테스트 집합에서 평가해요. 데이터를 준비하기 위해 두 개의 피처 변환기를 사용해요. 이들은 레이블과 범주형 피처의 범주를 인덱싱해 트리 기반 알고리즘이 인식할 수 있는 메타데이터를 DataFrame에 추가해요.

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml import Pipeline
from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.feature import IndexToString, StringIndexer, VectorIndexer
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# Load and parse the data file, converting it to a DataFrame.
data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

# Index labels, adding metadata to the label column.
# Fit on whole dataset to include all labels in index.
labelIndexer = StringIndexer(inputCol="label", outputCol="indexedLabel").fit(data)

# Automatically identify categorical features, and index them.
# Set maxCategories so features with > 4 distinct values are treated as continuous.
featureIndexer =\
    VectorIndexer(inputCol="features", outputCol="indexedFeatures", maxCategories=4).fit(data)

# Split the data into training and test sets (30% held out for testing)
(trainingData, testData) = data.randomSplit([0.7, 0.3])

# Train a RandomForest model.
rf = RandomForestClassifier(labelCol="indexedLabel", featuresCol="indexedFeatures", numTrees=10)

# Convert indexed labels back to original labels.
labelConverter = IndexToString(inputCol="prediction", outputCol="predictedLabel",
                               labels=labelIndexer.labels)

# Chain indexers and forest in a Pipeline
pipeline = Pipeline(stages=[labelIndexer, featureIndexer, rf, labelConverter])

# Train model.  This also runs the indexers.
model = pipeline.fit(trainingData)

# Make predictions.
predictions = model.transform(testData)

# Select example rows to display.
predictions.select("predictedLabel", "label", "features").show(5)

# Select (prediction, true label) and compute test error
evaluator = MulticlassClassificationEvaluator(
    labelCol="indexedLabel", predictionCol="prediction", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Test Error = %g" % (1.0 - accuracy))

rfModel = model.stages[2]
print(rfModel)  # summary only

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/random_forest_classifier_example.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.{RandomForestClassificationModel, RandomForestClassifier}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}

// Load and parse the data file, converting it to a DataFrame.
val data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

// Index labels, adding metadata to the label column.
// Fit on whole dataset to include all labels in index.
val labelIndexer = new StringIndexer()
  .setInputCol("label")
  .setOutputCol("indexedLabel")
  .fit(data)
// Automatically identify categorical features, and index them.
// Set maxCategories so features with > 4 distinct values are treated as continuous.
val featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4)
  .fit(data)

// Split the data into training and test sets (30% held out for testing).
val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))

// Train a RandomForest model.
val rf = new RandomForestClassifier()
  .setLabelCol("indexedLabel")
  .setFeaturesCol("indexedFeatures")
  .setNumTrees(10)

// Convert indexed labels back to original labels.
val labelConverter = new IndexToString()
  .setInputCol("prediction")
  .setOutputCol("predictedLabel")
  .setLabels(labelIndexer.labelsArray(0))

// Chain indexers and forest in a Pipeline.
val pipeline = new Pipeline()
  .setStages(Array(labelIndexer, featureIndexer, rf, labelConverter))

// Train model. This also runs the indexers.
val model = pipeline.fit(trainingData)

// Make predictions.
val predictions = model.transform(testData)

// Select example rows to display.
predictions.select("predictedLabel", "label", "features").show(5)

// Select (prediction, true label) and compute test error.
val evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("indexedLabel")
  .setPredictionCol("prediction")
  .setMetricName("accuracy")
val accuracy = evaluator.evaluate(predictions)
println(s"Test Error = ${(1.0 - accuracy)}")

val rfModel = model.stages(2).asInstanceOf[RandomForestClassificationModel]
println(s"Learned classification forest model:\n ${rfModel.toDebugString}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/RandomForestClassifierExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.RandomForestClassificationModel;
import org.apache.spark.ml.classification.RandomForestClassifier;
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
import org.apache.spark.ml.feature.*;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load and parse the data file, converting it to a DataFrame.
Dataset<Row> data = spark.read().format("libsvm").load("data/mllib/sample_libsvm_data.txt");

// Index labels, adding metadata to the label column.
// Fit on whole dataset to include all labels in index.
StringIndexerModel labelIndexer = new StringIndexer()
  .setInputCol("label")
  .setOutputCol("indexedLabel")
  .fit(data);
// Automatically identify categorical features, and index them.
// Set maxCategories so features with > 4 distinct values are treated as continuous.
VectorIndexerModel featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4)
  .fit(data);

// Split the data into training and test sets (30% held out for testing)
Dataset<Row>[] splits = data.randomSplit(new double[] {0.7, 0.3});
Dataset<Row> trainingData = splits[0];
Dataset<Row> testData = splits[1];

// Train a RandomForest model.
RandomForestClassifier rf = new RandomForestClassifier()
  .setLabelCol("indexedLabel")
  .setFeaturesCol("indexedFeatures");

// Convert indexed labels back to original labels.
IndexToString labelConverter = new IndexToString()
  .setInputCol("prediction")
  .setOutputCol("predictedLabel")
  .setLabels(labelIndexer.labelsArray()[0]);

// Chain indexers and forest in a Pipeline
Pipeline pipeline = new Pipeline()
  .setStages(new PipelineStage[] {labelIndexer, featureIndexer, rf, labelConverter});

// Train model. This also runs the indexers.
PipelineModel model = pipeline.fit(trainingData);

// Make predictions.
Dataset<Row> predictions = model.transform(testData);

// Select example rows to display.
predictions.select("predictedLabel", "label", "features").show(5);

// Select (prediction, true label) and compute test error
MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("indexedLabel")
  .setPredictionCol("prediction")
  .setMetricName("accuracy");
double accuracy = evaluator.evaluate(predictions);
System.out.println("Test Error = " + (1.0 - accuracy));

RandomForestClassificationModel rfModel = (RandomForestClassificationModel)(model.stages()[2]);
System.out.println("Learned classification forest model:\n" + rfModel.toDebugString());

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaRandomForestClassifierExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

# Load training data
df <- read.df("data/mllib/sample_libsvm_data.txt", source = "libsvm")
training <- df
test <- df

# Fit a random forest classification model with spark.randomForest
model <- spark.randomForest(training, label ~ features, "classification", numTrees = 10)

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/randomForest.R"에서 찾을 수 있어요.

그래디언트 부스티드 트리 분류기(Gradient-boosted tree classifier)

그래디언트 부스티드 트리(GBT)는 의사결정 트리 앙상블을 사용하는 인기 있는 분류·회귀 메서드예요. spark.ml 구현에 대한 더 많은 정보는 GBT 섹션에서 찾을 수 있어요.

예제

다음 예제는 LibSVM 형식의 데이터셋을 로드하고, 훈련·테스트 집합으로 나누고, 첫 데이터셋에서 훈련한 뒤 분리된 테스트 집합에서 평가해요. 데이터를 준비하기 위해 두 개의 피처 변환기를 사용해요. 이들은 트리 기반 알고리즘이 인식할 수 있는 메타데이터를 DataFrame에 추가해요.

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml import Pipeline
from pyspark.ml.classification import GBTClassifier
from pyspark.ml.feature import StringIndexer, VectorIndexer
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# Load and parse the data file, converting it to a DataFrame.
data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

# Index labels, adding metadata to the label column.
# Fit on whole dataset to include all labels in index.
labelIndexer = StringIndexer(inputCol="label", outputCol="indexedLabel").fit(data)
# Automatically identify categorical features, and index them.
# Set maxCategories so features with > 4 distinct values are treated as continuous.
featureIndexer =\
    VectorIndexer(inputCol="features", outputCol="indexedFeatures", maxCategories=4).fit(data)

# Split the data into training and test sets (30% held out for testing)
(trainingData, testData) = data.randomSplit([0.7, 0.3])

# Train a GBT model.
gbt = GBTClassifier(labelCol="indexedLabel", featuresCol="indexedFeatures", maxIter=10)

# Chain indexers and GBT in a Pipeline
pipeline = Pipeline(stages=[labelIndexer, featureIndexer, gbt])

# Train model.  This also runs the indexers.
model = pipeline.fit(trainingData)

# Make predictions.
predictions = model.transform(testData)

# Select example rows to display.
predictions.select("prediction", "indexedLabel", "features").show(5)

# Select (prediction, true label) and compute test error
evaluator = MulticlassClassificationEvaluator(
    labelCol="indexedLabel", predictionCol="prediction", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Test Error = %g" % (1.0 - accuracy))

gbtModel = model.stages[2]
print(gbtModel)  # summary only

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/gradient_boosted_tree_classifier_example.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.{GBTClassificationModel, GBTClassifier}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}

// Load and parse the data file, converting it to a DataFrame.
val data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

// Index labels, adding metadata to the label column.
// Fit on whole dataset to include all labels in index.
val labelIndexer = new StringIndexer()
  .setInputCol("label")
  .setOutputCol("indexedLabel")
  .fit(data)
// Automatically identify categorical features, and index them.
// Set maxCategories so features with > 4 distinct values are treated as continuous.
val featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4)
  .fit(data)

// Split the data into training and test sets (30% held out for testing).
val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))

// Train a GBT model.
val gbt = new GBTClassifier()
  .setLabelCol("indexedLabel")
  .setFeaturesCol("indexedFeatures")
  .setMaxIter(10)
  .setFeatureSubsetStrategy("auto")

// Convert indexed labels back to original labels.
val labelConverter = new IndexToString()
  .setInputCol("prediction")
  .setOutputCol("predictedLabel")
  .setLabels(labelIndexer.labelsArray(0))

// Chain indexers and GBT in a Pipeline.
val pipeline = new Pipeline()
  .setStages(Array(labelIndexer, featureIndexer, gbt, labelConverter))

// Train model. This also runs the indexers.
val model = pipeline.fit(trainingData)

// Make predictions.
val predictions = model.transform(testData)

// Select example rows to display.
predictions.select("predictedLabel", "label", "features").show(5)

// Select (prediction, true label) and compute test error.
val evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("indexedLabel")
  .setPredictionCol("prediction")
  .setMetricName("accuracy")
val accuracy = evaluator.evaluate(predictions)
println(s"Test Error = ${1.0 - accuracy}")

val gbtModel = model.stages(2).asInstanceOf[GBTClassificationModel]
println(s"Learned classification GBT model:\n ${gbtModel.toDebugString}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/GradientBoostedTreeClassifierExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.GBTClassificationModel;
import org.apache.spark.ml.classification.GBTClassifier;
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
import org.apache.spark.ml.feature.*;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load and parse the data file, converting it to a DataFrame.
Dataset<Row> data = spark
  .read()
  .format("libsvm")
  .load("data/mllib/sample_libsvm_data.txt");

// Index labels, adding metadata to the label column.
// Fit on whole dataset to include all labels in index.
StringIndexerModel labelIndexer = new StringIndexer()
  .setInputCol("label")
  .setOutputCol("indexedLabel")
  .fit(data);
// Automatically identify categorical features, and index them.
// Set maxCategories so features with > 4 distinct values are treated as continuous.
VectorIndexerModel featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4)
  .fit(data);

// Split the data into training and test sets (30% held out for testing)
Dataset<Row>[] splits = data.randomSplit(new double[] {0.7, 0.3});
Dataset<Row> trainingData = splits[0];
Dataset<Row> testData = splits[1];

// Train a GBT model.
GBTClassifier gbt = new GBTClassifier()
  .setLabelCol("indexedLabel")
  .setFeaturesCol("indexedFeatures")
  .setMaxIter(10);

// Convert indexed labels back to original labels.
IndexToString labelConverter = new IndexToString()
  .setInputCol("prediction")
  .setOutputCol("predictedLabel")
  .setLabels(labelIndexer.labelsArray()[0]);

// Chain indexers and GBT in a Pipeline.
Pipeline pipeline = new Pipeline()
  .setStages(new PipelineStage[] {labelIndexer, featureIndexer, gbt, labelConverter});

// Train model. This also runs the indexers.
PipelineModel model = pipeline.fit(trainingData);

// Make predictions.
Dataset<Row> predictions = model.transform(testData);

// Select example rows to display.
predictions.select("predictedLabel", "label", "features").show(5);

// Select (prediction, true label) and compute test error.
MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("indexedLabel")
  .setPredictionCol("prediction")
  .setMetricName("accuracy");
double accuracy = evaluator.evaluate(predictions);
System.out.println("Test Error = " + (1.0 - accuracy));

GBTClassificationModel gbtModel = (GBTClassificationModel)(model.stages()[2]);
System.out.println("Learned classification GBT model:\n" + gbtModel.toDebugString());

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaGradientBoostedTreeClassifierExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

# Load training data
df <- read.df("data/mllib/sample_libsvm_data.txt", source = "libsvm")
training <- df
test <- df

# Fit a GBT classification model with spark.gbt
model <- spark.gbt(training, label ~ features, "classification", maxIter = 10)

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/gbt.R"에서 찾을 수 있어요.

다층 퍼셉트론 분류기(Multilayer perceptron classifier)

다층 퍼셉트론 분류기(MLPC)는 피드포워드 인공 신경망에 기반한 분류기예요. MLPC는 여러 노드 층으로 구성돼요. 각 층은 네트워크의 다음 층에 완전히 연결돼요. 입력 층의 노드는 입력 데이터를 나타내요. 다른 모든 노드는 노드의 가중치 $\wv$와 편향 $\bv$로 입력의 선형 결합을 하고 활성화 함수를 적용해 입력을 출력으로 매핑해요. 이는 $K+1$개의 층을 가진 MLPC에 대해 행렬 형태로 다음과 같이 쓸 수 있어요.

\[ \mathrm{y}(\x) = \mathrm{f_K}(...\mathrm{f_2}(\wv_2^T\mathrm{f_1}(\wv_1^T \x+b_1)+b_2)...+b_K) \]

중간 층의 노드는 시그모이드(logistic) 함수를 사용해요.

\[ \mathrm{f}(z_i) = \frac{1}{1 + e^{-z_i}} \]

출력 층의 노드는 softmax 함수를 사용해요.

\[ \mathrm{f}(z_i) = \frac{e^{z_i}}{\sum_{k=1}^N e^{z_k}} \]

출력 층의 노드 수 $N$은 클래스 수에 해당해요.

MLPC는 모델 학습에 역전파(backpropagation)를 사용해요. 최적화에는 로지스틱 손실 함수를 사용하고 최적화 루틴으로 L-BFGS를 사용해요.

예제

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# Load training data
data = spark.read.format("libsvm")\
    .load("data/mllib/sample_multiclass_classification_data.txt")

# Split the data into train and test
splits = data.randomSplit([0.6, 0.4], 1234)
train = splits[0]
test = splits[1]

# specify layers for the neural network:
# input layer of size 4 (features), two intermediate of size 5 and 4
# and output of size 3 (classes)
layers = [4, 5, 4, 3]

# create the trainer and set its parameters
trainer = MultilayerPerceptronClassifier(maxIter=100, layers=layers, blockSize=128, seed=1234)

# train the model
model = trainer.fit(train)

# compute accuracy on the test set
result = model.transform(test)
predictionAndLabels = result.select("prediction", "label")
evaluator = MulticlassClassificationEvaluator(metricName="accuracy")
print("Test set accuracy = " + str(evaluator.evaluate(predictionAndLabels)))

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/multilayer_perceptron_classification.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.classification.MultilayerPerceptronClassifier
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator

// Load the data stored in LIBSVM format as a DataFrame.
val data = spark.read.format("libsvm")
  .load("data/mllib/sample_multiclass_classification_data.txt")

// Split the data into train and test
val splits = data.randomSplit(Array(0.6, 0.4), seed = 1234L)
val train = splits(0)
val test = splits(1)

// specify layers for the neural network:
// input layer of size 4 (features), two intermediate of size 5 and 4
// and output of size 3 (classes)
val layers = Array[Int](4, 5, 4, 3)

// create the trainer and set its parameters
val trainer = new MultilayerPerceptronClassifier()
  .setLayers(layers)
  .setBlockSize(128)
  .setSeed(1234L)
  .setMaxIter(100)

// train the model
val model = trainer.fit(train)

// compute accuracy on the test set
val result = model.transform(test)
val predictionAndLabels = result.select("prediction", "label")
val evaluator = new MulticlassClassificationEvaluator()
  .setMetricName("accuracy")

println(s"Test set accuracy = ${evaluator.evaluate(predictionAndLabels)}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/MultilayerPerceptronClassifierExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.ml.classification.MultilayerPerceptronClassificationModel;
import org.apache.spark.ml.classification.MultilayerPerceptronClassifier;
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;

// Load training data
String path = "data/mllib/sample_multiclass_classification_data.txt";
Dataset<Row> dataFrame = spark.read().format("libsvm").load(path);

// Split the data into train and test
Dataset<Row>[] splits = dataFrame.randomSplit(new double[]{0.6, 0.4}, 1234L);
Dataset<Row> train = splits[0];
Dataset<Row> test = splits[1];

// specify layers for the neural network:
// input layer of size 4 (features), two intermediate of size 5 and 4
// and output of size 3 (classes)
int[] layers = new int[] {4, 5, 4, 3};

// create the trainer and set its parameters
MultilayerPerceptronClassifier trainer = new MultilayerPerceptronClassifier()
  .setLayers(layers)
  .setBlockSize(128)
  .setSeed(1234L)
  .setMaxIter(100);

// train the model
MultilayerPerceptronClassificationModel model = trainer.fit(train);

// compute accuracy on the test set
Dataset<Row> result = model.transform(test);
Dataset<Row> predictionAndLabels = result.select("prediction", "label");
MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
  .setMetricName("accuracy");

System.out.println("Test set accuracy = " + evaluator.evaluate(predictionAndLabels));

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaMultilayerPerceptronClassifierExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

# Load training data
df <- read.df("data/mllib/sample_multiclass_classification_data.txt", source = "libsvm")
training <- df
test <- df

# specify layers for the neural network:
# input layer of size 4 (features), two intermediate of size 5 and 4
# and output of size 3 (classes)
layers = c(4, 5, 4, 3)

# Fit a multi-layer perceptron neural network model with spark.mlp
model <- spark.mlp(training, label ~ features, maxIter = 100,
                   layers = layers, blockSize = 128, seed = 1234)

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/mlp.R"에서 찾을 수 있어요.

선형 서포트 벡터 머신(Linear Support Vector Machine)

서포트 벡터 머신은 고차원 또는 무한 차원 공간에서 초평면 또는 초평면 집합을 구성하는데, 분류, 회귀, 기타 작업에 쓸 수 있어요. 직관적으로 좋은 분리는 어떤 클래스의 가장 가까운 훈련 데이터 포인트까지의 거리(소위 functional margin)가 가장 큰 초평면으로 달성돼요. 일반적으로 마진이 클수록 분류기의 일반화 오차가 낮아지기 때문이에요. Spark ML의 LinearSVC는 선형 SVM으로 이진 분류를 지원해요. 내부적으로 OWLQN 옵티마이저로 Hinge Loss를 최적화해요.

예제

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml.classification import LinearSVC

# Load training data
training = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

lsvc = LinearSVC(maxIter=10, regParam=0.1)

# Fit the model
lsvcModel = lsvc.fit(training)

# Print the coefficients and intercept for linear SVC
print("Coefficients: " + str(lsvcModel.coefficients))
print("Intercept: " + str(lsvcModel.intercept))

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/linearsvc.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.classification.LinearSVC

// Load training data
val training = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

val lsvc = new LinearSVC()
  .setMaxIter(10)
  .setRegParam(0.1)

// Fit the model
val lsvcModel = lsvc.fit(training)

// Print the coefficients and intercept for linear svc
println(s"Coefficients: ${lsvcModel.coefficients} Intercept: ${lsvcModel.intercept}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/LinearSVCExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import org.apache.spark.ml.classification.LinearSVC;
import org.apache.spark.ml.classification.LinearSVCModel;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load training data
Dataset<Row> training = spark.read().format("libsvm")
  .load("data/mllib/sample_libsvm_data.txt");

LinearSVC lsvc = new LinearSVC()
  .setMaxIter(10)
  .setRegParam(0.1);

// Fit the model
LinearSVCModel lsvcModel = lsvc.fit(training);

// Print the coefficients and intercept for LinearSVC
System.out.println("Coefficients: "
  + lsvcModel.coefficients() + " Intercept: " + lsvcModel.intercept());

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaLinearSVCExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

# load training data
t <- as.data.frame(Titanic)
training <- createDataFrame(t)

# fit Linear SVM model
model <- spark.svmLinear(training,  Survived ~ ., regParam = 0.01, maxIter = 10)

# Model summary
summary(model)

# Prediction
prediction <- predict(model, training)
showDF(prediction)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/svmLinear.R"에서 찾을 수 있어요.

One-vs-Rest 분류기(One-vs-All)

OneVsRest는 이진 분류를 효율적으로 수행할 수 있는 기본 분류기가 주어졌을 때 다중 클래스 분류를 수행하는 머신러닝 환원(reduction)의 예시예요. "One-vs-All"이라고도 알려져 있어요.

OneVsRest는 Estimator로 구현돼요. 기본 분류기로 Classifier 인스턴스를 받아 k개의 각 클래스에 대한 이진 분류 문제를 만들어요. 클래스 i의 분류기는 레이블이 i인지 아닌지를 예측하도록 훈련되어, 클래스 i를 다른 모든 클래스와 구분해요.

예측은 각 이진 분류기를 평가해 이루어지며, 가장 확신 있는 분류기의 인덱스가 레이블로 출력돼요.

예제

아래 예제는 Iris 데이터셋을 로드하고, DataFrame으로 파싱한 뒤 OneVsRest로 다중 클래스 분류를 수행하는 방법을 보여줘요. 알고리즘 정확도를 측정하기 위해 테스트 오차가 계산돼요.

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml.classification import LogisticRegression, OneVsRest
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# load data file.
inputData = spark.read.format("libsvm") \
    .load("data/mllib/sample_multiclass_classification_data.txt")

# generate the train/test split.
(train, test) = inputData.randomSplit([0.8, 0.2])

# instantiate the base classifier.
lr = LogisticRegression(maxIter=10, tol=1E-6, fitIntercept=True)

# instantiate the One Vs Rest Classifier.
ovr = OneVsRest(classifier=lr)

# train the multiclass model.
ovrModel = ovr.fit(train)

# score the model on test data.
predictions = ovrModel.transform(test)

# obtain evaluator.
evaluator = MulticlassClassificationEvaluator(metricName="accuracy")

# compute the classification error on test data.
accuracy = evaluator.evaluate(predictions)
print("Test Error = %g" % (1.0 - accuracy))

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/one_vs_rest_example.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.classification.{LogisticRegression, OneVsRest}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator

// load data file.
val inputData = spark.read.format("libsvm")
  .load("data/mllib/sample_multiclass_classification_data.txt")

// generate the train/test split.
val Array(train, test) = inputData.randomSplit(Array(0.8, 0.2))

// instantiate the base classifier
val classifier = new LogisticRegression()
  .setMaxIter(10)
  .setTol(1E-6)
  .setFitIntercept(true)

// instantiate the One Vs Rest Classifier.
val ovr = new OneVsRest().setClassifier(classifier)

// train the multiclass model.
val ovrModel = ovr.fit(train)

// score the model on test data.
val predictions = ovrModel.transform(test)

// obtain evaluator.
val evaluator = new MulticlassClassificationEvaluator()
  .setMetricName("accuracy")

// compute the classification error on test data.
val accuracy = evaluator.evaluate(predictions)
println(s"Test Error = ${1 - accuracy}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/OneVsRestExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import org.apache.spark.ml.classification.LogisticRegression;
import org.apache.spark.ml.classification.OneVsRest;
import org.apache.spark.ml.classification.OneVsRestModel;
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

// load data file.
Dataset<Row> inputData = spark.read().format("libsvm")
  .load("data/mllib/sample_multiclass_classification_data.txt");

// generate the train/test split.
Dataset<Row>[] tmp = inputData.randomSplit(new double[]{0.8, 0.2});
Dataset<Row> train = tmp[0];
Dataset<Row> test = tmp[1];

// configure the base classifier.
LogisticRegression classifier = new LogisticRegression()
  .setMaxIter(10)
  .setTol(1E-6)
  .setFitIntercept(true);

// instantiate the One Vs Rest Classifier.
OneVsRest ovr = new OneVsRest().setClassifier(classifier);

// train the multiclass model.
OneVsRestModel ovrModel = ovr.fit(train);

// score the model on test data.
Dataset<Row> predictions = ovrModel.transform(test)
  .select("prediction", "label");

// obtain evaluator.
MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
        .setMetricName("accuracy");

// compute the classification error on test data.
double accuracy = evaluator.evaluate(predictions);
System.out.println("Test Error = " + (1 - accuracy));

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaOneVsRestExample.java"에서 찾을 수 있어요.

나이브 베이즈(Naive Bayes)

나이브 베이즈 분류기는 모든 피처 쌍 사이에 강한(순진한) 독립 가정으로 베이즈 정리를 적용하는 간단한 확률적 다중 클래스 분류기 패밀리예요.

나이브 베이즈는 매우 효율적으로 훈련될 수 있어요. 훈련 데이터를 한 번만 순회해 각 피처가 각 레이블에 주어졌을 때의 조건부 확률 분포를 계산해요. 예측을 위해 베이즈 정리를 적용해 관측치가 주어졌을 때 각 레이블의 조건부 확률 분포를 계산해요.

MLlib는 Multinomial naive Bayes, Complement naive Bayes, Bernoulli naive Bayes, Gaussian naive Bayes를 지원해요.

입력 데이터: 이 Multinomial, Complement, Bernoulli 모델은 보통 문서 분류에 사용돼요. 그 맥락에서 각 관측치는 문서이고 각 피처는 용어를 나타내요. 피처의 값은 용어의 빈도(Multinomial이나 Complement 나이브 베이즈에서) 또는 문서에서 용어가 발견됐는지 여부를 나타내는 0 또는 1(Bernoulli 나이브 베이즈에서)이에요. Multinomial과 Bernoulli 모델의 피처 값은 음수가 아니어야 해요. 모델 타입은 선택적 파라미터 "multinomial", "complement", "bernoulli" 또는 "gaussian"으로 선택하며 기본은 "multinomial"이에요. 문서 분류의 경우 입력 피처 벡터는 보통 희소 벡터여야 해요. 훈련 데이터는 한 번만 사용되므로 캐시할 필요가 없어요.

가산 평활(Additive smoothing)은 파라미터 $\lambda$(기본 $1.0$)를 설정해 쓸 수 있어요.

예제

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml.classification import NaiveBayes
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# Load training data
data = spark.read.format("libsvm") \
    .load("data/mllib/sample_libsvm_data.txt")

# Split the data into train and test
splits = data.randomSplit([0.6, 0.4], 1234)
train = splits[0]
test = splits[1]

# create the trainer and set its parameters
nb = NaiveBayes(smoothing=1.0, modelType="multinomial")

# train the model
model = nb.fit(train)

# select example rows to display.
predictions = model.transform(test)
predictions.show()

# compute accuracy on the test set
evaluator = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction",
                                              metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Test set accuracy = " + str(accuracy))

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/naive_bayes_example.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.classification.NaiveBayes
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator

// Load the data stored in LIBSVM format as a DataFrame.
val data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

// Split the data into training and test sets (30% held out for testing)
val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3), seed = 1234L)

// Train a NaiveBayes model.
val model = new NaiveBayes()
  .fit(trainingData)

// Select example rows to display.
val predictions = model.transform(testData)
predictions.show()

// Select (prediction, true label) and compute test error
val evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("accuracy")
val accuracy = evaluator.evaluate(predictions)
println(s"Test set accuracy = $accuracy")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/NaiveBayesExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import org.apache.spark.ml.classification.NaiveBayes;
import org.apache.spark.ml.classification.NaiveBayesModel;
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load training data
Dataset<Row> dataFrame =
  spark.read().format("libsvm").load("data/mllib/sample_libsvm_data.txt");
// Split the data into train and test
Dataset<Row>[] splits = dataFrame.randomSplit(new double[]{0.6, 0.4}, 1234L);
Dataset<Row> train = splits[0];
Dataset<Row> test = splits[1];

// create the trainer and set its parameters
NaiveBayes nb = new NaiveBayes();

// train the model
NaiveBayesModel model = nb.fit(train);

// Select example rows to display.
Dataset<Row> predictions = model.transform(test);
predictions.show();

// compute accuracy on the test set
MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("accuracy");
double accuracy = evaluator.evaluate(predictions);
System.out.println("Test set accuracy = " + accuracy);

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaNaiveBayesExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

# Fit a Bernoulli naive Bayes model with spark.naiveBayes
titanic <- as.data.frame(Titanic)
titanicDF <- createDataFrame(titanic[titanic$Freq > 0, -5])
nbDF <- titanicDF
nbTestDF <- titanicDF
nbModel <- spark.naiveBayes(nbDF, Survived ~ Class + Sex + Age)

# Model summary
summary(nbModel)

# Prediction
nbPredictions <- predict(nbModel, nbTestDF)
head(nbPredictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/naiveBayes.R"에서 찾을 수 있어요.

팩터라이제이션 머신 분류기(Factorization machines classifier)

팩터라이제이션 머신 구현에 대한 더 많은 배경과 세부 사항은 Factorization Machines 섹션을 참고하세요.

예제

다음 예제는 LibSVM 형식의 데이터셋을 로드하고, 훈련·테스트 집합으로 나누고, 첫 데이터셋에서 훈련한 뒤 분리된 테스트 집합에서 평가해요. 폭주하는 그라디언트(exploding gradient) 문제를 막기 위해 피처를 0과 1 사이로 스케일링해요.

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml import Pipeline
from pyspark.ml.classification import FMClassifier
from pyspark.ml.feature import MinMaxScaler, StringIndexer
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# Load and parse the data file, converting it to a DataFrame.
data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

# Index labels, adding metadata to the label column.
# Fit on whole dataset to include all labels in index.
labelIndexer = StringIndexer(inputCol="label", outputCol="indexedLabel").fit(data)
# Scale features.
featureScaler = MinMaxScaler(inputCol="features", outputCol="scaledFeatures").fit(data)

# Split the data into training and test sets (30% held out for testing)
(trainingData, testData) = data.randomSplit([0.7, 0.3])

# Train a FM model.
fm = FMClassifier(labelCol="indexedLabel", featuresCol="scaledFeatures", stepSize=0.001)

# Create a Pipeline.
pipeline = Pipeline(stages=[labelIndexer, featureScaler, fm])

# Train model.
model = pipeline.fit(trainingData)

# Make predictions.
predictions = model.transform(testData)

# Select example rows to display.
predictions.select("prediction", "indexedLabel", "features").show(5)

# Select (prediction, true label) and compute test accuracy
evaluator = MulticlassClassificationEvaluator(
    labelCol="indexedLabel", predictionCol="prediction", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Test set accuracy = %g" % accuracy)

fmModel = model.stages[2]
print("Factors: " + str(fmModel.factors))  # type: ignore
print("Linear: " + str(fmModel.linear))  # type: ignore
print("Intercept: " + str(fmModel.intercept))  # type: ignore

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/fm_classifier_example.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.{FMClassificationModel, FMClassifier}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{IndexToString, MinMaxScaler, StringIndexer}

// Load and parse the data file, converting it to a DataFrame.
val data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

// Index labels, adding metadata to the label column.
// Fit on whole dataset to include all labels in index.
val labelIndexer = new StringIndexer()
  .setInputCol("label")
  .setOutputCol("indexedLabel")
  .fit(data)
// Scale features.
val featureScaler = new MinMaxScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")
  .fit(data)

// Split the data into training and test sets (30% held out for testing).
val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))

// Train a FM model.
val fm = new FMClassifier()
  .setLabelCol("indexedLabel")
  .setFeaturesCol("scaledFeatures")
  .setStepSize(0.001)

// Convert indexed labels back to original labels.
val labelConverter = new IndexToString()
  .setInputCol("prediction")
  .setOutputCol("predictedLabel")
  .setLabels(labelIndexer.labelsArray(0))

// Create a Pipeline.
val pipeline = new Pipeline()
  .setStages(Array(labelIndexer, featureScaler, fm, labelConverter))

// Train model.
val model = pipeline.fit(trainingData)

// Make predictions.
val predictions = model.transform(testData)

// Select example rows to display.
predictions.select("predictedLabel", "label", "features").show(5)

// Select (prediction, true label) and compute test accuracy.
val evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("indexedLabel")
  .setPredictionCol("prediction")
  .setMetricName("accuracy")
val accuracy = evaluator.evaluate(predictions)
println(s"Test set accuracy = $accuracy")

val fmModel = model.stages(2).asInstanceOf[FMClassificationModel]
println(s"Factors: ${fmModel.factors} Linear: ${fmModel.linear} " +
  s"Intercept: ${fmModel.intercept}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/FMClassifierExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.FMClassificationModel;
import org.apache.spark.ml.classification.FMClassifier;
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
import org.apache.spark.ml.feature.*;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load and parse the data file, converting it to a DataFrame.
Dataset<Row> data = spark
    .read()
    .format("libsvm")
    .load("data/mllib/sample_libsvm_data.txt");

// Index labels, adding metadata to the label column.
// Fit on whole dataset to include all labels in index.
StringIndexerModel labelIndexer = new StringIndexer()
    .setInputCol("label")
    .setOutputCol("indexedLabel")
    .fit(data);
// Scale features.
MinMaxScalerModel featureScaler = new MinMaxScaler()
    .setInputCol("features")
    .setOutputCol("scaledFeatures")
    .fit(data);

// Split the data into training and test sets (30% held out for testing)
Dataset<Row>[] splits = data.randomSplit(new double[] {0.7, 0.3});
Dataset<Row> trainingData = splits[0];
Dataset<Row> testData = splits[1];

// Train a FM model.
FMClassifier fm = new FMClassifier()
    .setLabelCol("indexedLabel")
    .setFeaturesCol("scaledFeatures")
    .setStepSize(0.001);

// Convert indexed labels back to original labels.
IndexToString labelConverter = new IndexToString()
    .setInputCol("prediction")
    .setOutputCol("predictedLabel")
    .setLabels(labelIndexer.labelsArray()[0]);

// Create a Pipeline.
Pipeline pipeline = new Pipeline()
    .setStages(new PipelineStage[] {labelIndexer, featureScaler, fm, labelConverter});

// Train model.
PipelineModel model = pipeline.fit(trainingData);

// Make predictions.
Dataset<Row> predictions = model.transform(testData);

// Select example rows to display.
predictions.select("predictedLabel", "label", "features").show(5);

// Select (prediction, true label) and compute test accuracy.
MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
    .setLabelCol("indexedLabel")
    .setPredictionCol("prediction")
    .setMetricName("accuracy");
double accuracy = evaluator.evaluate(predictions);
System.out.println("Test Accuracy = " + accuracy);

FMClassificationModel fmModel = (FMClassificationModel)(model.stages()[2]);
System.out.println("Factors: " + fmModel.factors());
System.out.println("Linear: " + fmModel.linear());
System.out.println("Intercept: " + fmModel.intercept());

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaFMClassifierExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

참고: 현재 SparkR은 피처 스케일링을 지원하지 않아요.

# Load training data
df <- read.df("data/mllib/sample_libsvm_data.txt", source = "libsvm")
training <- df
test <- df

# Fit a FM classification model
model <- spark.fmClassifier(training, label ~ features)

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/fmClassifier.R"에서 찾을 수 있어요.

회귀(Regression)

선형 회귀(Linear regression)

선형 회귀 모델과 모델 요약을 다루는 인터페이스는 로지스틱 회귀 경우와 비슷해요.

"l-bfgs" 솔버로 절편 없이 상수가 아닌 0이 아닌 컬럼이 있는 데이터셋에서 LinearRegressionModel을 피팅할 때, Spark MLlib는 상수 0이 아닌 컬럼에 대해 0 계수를 출력해요. 이 동작은 R glmnet과 같지만 LIBSVM과는 달라요.

예제

다음 예제는 elastic net 정규화된 선형 회귀 모델을 훈련하고 모델 요약 통계를 추출하는 방법을 보여줘요.

파라미터에 대한 자세한 내용은 Python API 문서에서 찾을 수 있어요.

from pyspark.ml.regression import LinearRegression

# Load training data
training = spark.read.format("libsvm")\
    .load("data/mllib/sample_linear_regression_data.txt")

lr = LinearRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8)

# Fit the model
lrModel = lr.fit(training)

# Print the coefficients and intercept for linear regression
print("Coefficients: %s" % str(lrModel.coefficients))
print("Intercept: %s" % str(lrModel.intercept))

# Summarize the model over the training set and print out some metrics
trainingSummary = lrModel.summary
print("numIterations: %d" % trainingSummary.totalIterations)
print("objectiveHistory: %s" % str(trainingSummary.objectiveHistory))
trainingSummary.residuals.show()
print("RMSE: %f" % trainingSummary.rootMeanSquaredError)
print("r2: %f" % trainingSummary.r2)

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/linear_regression_with_elastic_net.py"에서 찾을 수 있어요.

파라미터에 대한 자세한 내용은 Scala API 문서에서 찾을 수 있어요.

import org.apache.spark.ml.regression.LinearRegression

// Load training data
val training = spark.read.format("libsvm")
  .load("data/mllib/sample_linear_regression_data.txt")

val lr = new LinearRegression()
  .setMaxIter(10)
  .setRegParam(0.3)
  .setElasticNetParam(0.8)

// Fit the model
val lrModel = lr.fit(training)

// Print the coefficients and intercept for linear regression
println(s"Coefficients: ${lrModel.coefficients} Intercept: ${lrModel.intercept}")

// Summarize the model over the training set and print out some metrics
val trainingSummary = lrModel.summary
println(s"numIterations: ${trainingSummary.totalIterations}")
println(s"objectiveHistory: [${trainingSummary.objectiveHistory.mkString(",")}]")
trainingSummary.residuals.show()
println(s"RMSE: ${trainingSummary.rootMeanSquaredError}")
println(s"r2: ${trainingSummary.r2}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/LinearRegressionWithElasticNetExample.scala"에서 찾을 수 있어요.

파라미터에 대한 자세한 내용은 Java API 문서에서 찾을 수 있어요.

import org.apache.spark.ml.regression.LinearRegression;
import org.apache.spark.ml.regression.LinearRegressionModel;
import org.apache.spark.ml.regression.LinearRegressionTrainingSummary;
import org.apache.spark.ml.linalg.Vectors;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load training data.
Dataset<Row> training = spark.read().format("libsvm")
  .load("data/mllib/sample_linear_regression_data.txt");

LinearRegression lr = new LinearRegression()
  .setMaxIter(10)
  .setRegParam(0.3)
  .setElasticNetParam(0.8);

// Fit the model.
LinearRegressionModel lrModel = lr.fit(training);

// Print the coefficients and intercept for linear regression.
System.out.println("Coefficients: "
  + lrModel.coefficients() + " Intercept: " + lrModel.intercept());

// Summarize the model over the training set and print out some metrics.
LinearRegressionTrainingSummary trainingSummary = lrModel.summary();
System.out.println("numIterations: " + trainingSummary.totalIterations());
System.out.println("objectiveHistory: " + Vectors.dense(trainingSummary.objectiveHistory()));
trainingSummary.residuals().show();
System.out.println("RMSE: " + trainingSummary.rootMeanSquaredError());
System.out.println("r2: " + trainingSummary.r2());

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaLinearRegressionWithElasticNetExample.java"에서 찾을 수 있어요.

파라미터에 대한 자세한 내용은 R API 문서에서 찾을 수 있어요.

# Load training data
df <- read.df("data/mllib/sample_linear_regression_data.txt", source = "libsvm")
training <- df
test <- df

# Fit a linear regression model
model <- spark.lm(training, label ~ features, regParam = 0.3, elasticNetParam = 0.8)

# Prediction
predictions <- predict(model, test)
head(predictions)

# Summarize
summary(model)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/lm_with_elastic_net.R"에서 찾을 수 있어요.

일반화 선형 회귀(Generalized linear regression)

출력이 가우시안 분포를 따른다고 가정하는 선형 회귀와 대조적으로, 일반화 선형 모델(GLM)은 응답 변수 $Y_i$가 지수족 분포 중 어떤 분포를 따르는 선형 모델의 명세예요. Spark의 GeneralizedLinearRegression 인터페이스는 선형 회귀, 포아송 회귀, 로지스틱 회귀 등을 포함한 다양한 예측 문제에 쓸 수 있는 GLM의 유연한 명세를 허용해요. 현재 spark.ml에서는 지수족 분포의 일부만 지원되며 아래에 나열돼 있어요.

참고: Spark은 현재 GeneralizedLinearRegression 인터페이스를 통해 최대 4096개의 피처만 지원하며, 이 제약을 초과하면 예외를 던져요. 자세한 내용은 advanced section을 참고하세요. 그래도 선형·로지스틱 회귀의 경우 LinearRegression과 LogisticRegression 추정기를 사용해 더 많은 피처를 가진 모델을 훈련할 수 있어요.

GLM은 "표준" 또는 "자연" 형태로 쓸 수 있는 지수족 분포를 요구해요. 즉 자연 지수족 분포예요. 자연 지수족 분포의 형태는 다음과 같아요.

\[f_Y(y|\theta, \tau) = h(y, \tau)\exp{\left( \frac{\theta \cdot y - A(\theta)}{d(\tau)} \right)}\]

여기서 $\theta$는 관심 파라미터이고 $\tau$는 분산(dispersion) 파라미터예요. GLM에서 응답 변수 $Y_i$는 자연 지수족 분포에서 추출된다고 가정해요.

\[Y_i \sim f\left(\cdot|\theta_i, \tau \right)\]

여기서 관심 파라미터 $\theta_i$는 응답 변수의 기대값 $\mu_i$와 다음으로 연결돼요.

\[\mu_i = A'(\theta_i)\]

여기서 $A'(\theta_i)$는 선택한 분포의 형태로 정의돼요. GLM은 또한 응답 변수의 기대값 $\mu_i$와 소위 선형 예측자(linear predictor) $\eta_i$ 사이의 관계를 정의하는 링크 함수(link function)의 명세도 허용해요.

\[g(\mu_i) = \eta_i = \vec{x_i}^T \cdot \vec{\beta}\]

흔히 링크 함수는 $A' = g^{-1}$가 되도록 선택되는데, 이는 관심 파라미터 $\theta$와 선형 예측자 $\eta$ 사이의 관계를 단순화해요. 이 경우 링크 함수 $g(\mu)$는 "정규(canonical)" 링크 함수라고 해요.

\[\theta_i = A'^{-1}(\mu_i) = g(g^{-1}(\eta_i)) = \eta_i\]

GLM은 가능도 함수를 최대화하는 회귀 계수 $\vec{\beta}$를 찾아요.

\[\max_{\vec{\beta}} \mathcal{L}(\vec{\theta}|\vec{y},X) = \prod_{i=1}^{N} h(y_i, \tau) \exp{\left(\frac{y_i\theta_i - A(\theta_i)}{d(\tau)}\right)}\]

여기서 관심 파라미터 $\theta_i$는 회귀 계수 $\vec{\beta}$와 다음으로 연결돼요.

\[\theta_i = A'^{-1}(g^{-1}(\vec{x_i} \cdot \vec{\beta}))\]

Spark의 일반화 선형 회귀 인터페이스는 또한 잔차, p-값, 편차(deviance), Akaike 정보 기준 등을 포함해 GLM 모델 피팅을 진단하는 요약 통계를 제공해요.

GLM과 그 응용에 대한 더 포괄적인 검토는 여기를 참고하세요.

사용 가능한 패밀리(Available families)

패밀리	응답 타입	지원 링크
Gaussian	연속	Identity*, Log, Inverse
Binomial	이진	Logit*, Probit, CLogLog
Poisson	카운트	Log*, Identity, Sqrt
Gamma	연속	Inverse*, Identity, Log
Tweedie	영-과잉 연속	Power link function
* 정규 링크

예제

다음 예제는 가우시안 응답과 identity 링크 함수로 GLM을 훈련하고 모델 요약 통계를 추출하는 방법을 보여줘요.

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml.regression import GeneralizedLinearRegression

# Load training data
dataset = spark.read.format("libsvm")\
    .load("data/mllib/sample_linear_regression_data.txt")

glr = GeneralizedLinearRegression(family="gaussian", link="identity", maxIter=10, regParam=0.3)

# Fit the model
model = glr.fit(dataset)

# Print the coefficients and intercept for generalized linear regression model
print("Coefficients: " + str(model.coefficients))
print("Intercept: " + str(model.intercept))

# Summarize the model over the training set and print out some metrics
summary = model.summary
print("Coefficient Standard Errors: " + str(summary.coefficientStandardErrors))
print("T Values: " + str(summary.tValues))
print("P Values: " + str(summary.pValues))
print("Dispersion: " + str(summary.dispersion))
print("Null Deviance: " + str(summary.nullDeviance))
print("Residual Degree Of Freedom Null: " + str(summary.residualDegreeOfFreedomNull))
print("Deviance: " + str(summary.deviance))
print("Residual Degree Of Freedom: " + str(summary.residualDegreeOfFreedom))
print("AIC: " + str(summary.aic))
print("Deviance Residuals: ")
summary.residuals().show()

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/generalized_linear_regression_example.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.regression.GeneralizedLinearRegression

// Load training data
val dataset = spark.read.format("libsvm")
  .load("data/mllib/sample_linear_regression_data.txt")

val glr = new GeneralizedLinearRegression()
  .setFamily("gaussian")
  .setLink("identity")
  .setMaxIter(10)
  .setRegParam(0.3)

// Fit the model
val model = glr.fit(dataset)

// Print the coefficients and intercept for generalized linear regression model
println(s"Coefficients: ${model.coefficients}")
println(s"Intercept: ${model.intercept}")

// Summarize the model over the training set and print out some metrics
val summary = model.summary
println(s"Coefficient Standard Errors: ${summary.coefficientStandardErrors.mkString(",")}")
println(s"T Values: ${summary.tValues.mkString(",")}")
println(s"P Values: ${summary.pValues.mkString(",")}")
println(s"Dispersion: ${summary.dispersion}")
println(s"Null Deviance: ${summary.nullDeviance}")
println(s"Residual Degree Of Freedom Null: ${summary.residualDegreeOfFreedomNull}")
println(s"Deviance: ${summary.deviance}")
println(s"Residual Degree Of Freedom: ${summary.residualDegreeOfFreedom}")
println(s"AIC: ${summary.aic}")
println("Deviance Residuals: ")
summary.residuals().show()

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/GeneralizedLinearRegressionExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import java.util.Arrays;

import org.apache.spark.ml.regression.GeneralizedLinearRegression;
import org.apache.spark.ml.regression.GeneralizedLinearRegressionModel;
import org.apache.spark.ml.regression.GeneralizedLinearRegressionTrainingSummary;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

// Load training data
Dataset<Row> dataset = spark.read().format("libsvm")
  .load("data/mllib/sample_linear_regression_data.txt");

GeneralizedLinearRegression glr = new GeneralizedLinearRegression()
  .setFamily("gaussian")
  .setLink("identity")
  .setMaxIter(10)
  .setRegParam(0.3);

// Fit the model
GeneralizedLinearRegressionModel model = glr.fit(dataset);

// Print the coefficients and intercept for generalized linear regression model
System.out.println("Coefficients: " + model.coefficients());
System.out.println("Intercept: " + model.intercept());

// Summarize the model over the training set and print out some metrics
GeneralizedLinearRegressionTrainingSummary summary = model.summary();
System.out.println("Coefficient Standard Errors: "
  + Arrays.toString(summary.coefficientStandardErrors()));
System.out.println("T Values: " + Arrays.toString(summary.tValues()));
System.out.println("P Values: " + Arrays.toString(summary.pValues()));
System.out.println("Dispersion: " + summary.dispersion());
System.out.println("Null Deviance: " + summary.nullDeviance());
System.out.println("Residual Degree Of Freedom Null: " + summary.residualDegreeOfFreedomNull());
System.out.println("Deviance: " + summary.deviance());
System.out.println("Residual Degree Of Freedom: " + summary.residualDegreeOfFreedom());
System.out.println("AIC: " + summary.aic());
System.out.println("Deviance Residuals: ");
summary.residuals().show();

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaGeneralizedLinearRegressionExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

training <- read.df("data/mllib/sample_multiclass_classification_data.txt", source = "libsvm")
# Fit a generalized linear model of family "gaussian" with spark.glm
df_list <- randomSplit(training, c(7, 3), 2)
gaussianDF <- df_list[[1]]
gaussianTestDF <- df_list[[2]]
gaussianGLM <- spark.glm(gaussianDF, label ~ features, family = "gaussian")

# Model summary
summary(gaussianGLM)

# Prediction
gaussianPredictions <- predict(gaussianGLM, gaussianTestDF)
head(gaussianPredictions)

# Fit a generalized linear model with glm (R-compliant)
gaussianGLM2 <- glm(label ~ features, gaussianDF, family = "gaussian")
summary(gaussianGLM2)

# Fit a generalized linear model of family "binomial" with spark.glm
training2 <- read.df("data/mllib/sample_multiclass_classification_data.txt", source = "libsvm")
training2 <- transform(training2, label = cast(training2$label > 1, "integer"))
df_list2 <- randomSplit(training2, c(7, 3), 2)
binomialDF <- df_list2[[1]]
binomialTestDF <- df_list2[[2]]
binomialGLM <- spark.glm(binomialDF, label ~ features, family = "binomial")

# Model summary
summary(binomialGLM)

# Prediction
binomialPredictions <- predict(binomialGLM, binomialTestDF)
head(binomialPredictions)

# Fit a generalized linear model of family "tweedie" with spark.glm
training3 <- read.df("data/mllib/sample_multiclass_classification_data.txt", source = "libsvm")
tweedieDF <- transform(training3, label = training3$label * exp(randn(10)))
tweedieGLM <- spark.glm(tweedieDF, label ~ features, family = "tweedie",
                        var.power = 1.2, link.power = 0)

# Model summary
summary(tweedieGLM)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/glm.R"에서 찾을 수 있어요.

의사결정 트리 회귀(Decision tree regression)

의사결정 트리는 인기 있는 분류·회귀 메서드 패밀리예요. spark.ml 구현에 대한 더 많은 정보는 의사결정 트리 섹션에서 찾을 수 있어요.

예제

다음 예제는 LibSVM 형식의 데이터셋을 로드하고, 훈련·테스트 집합으로 나누고, 첫 데이터셋에서 훈련한 뒤 분리된 테스트 집합에서 평가해요. 의사결정 트리 알고리즘이 인식할 수 있는 메타데이터를 DataFrame에 추가하는 피처 변환기를 사용해 범주형 피처를 인덱싱해요.

파라미터에 대한 자세한 내용은 Python API 문서에서 찾을 수 있어요.

from pyspark.ml import Pipeline
from pyspark.ml.regression import DecisionTreeRegressor
from pyspark.ml.feature import VectorIndexer
from pyspark.ml.evaluation import RegressionEvaluator

# Load the data stored in LIBSVM format as a DataFrame.
data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

# Automatically identify categorical features, and index them.
# We specify maxCategories so features with > 4 distinct values are treated as continuous.
featureIndexer =\
    VectorIndexer(inputCol="features", outputCol="indexedFeatures", maxCategories=4).fit(data)

# Split the data into training and test sets (30% held out for testing)
(trainingData, testData) = data.randomSplit([0.7, 0.3])

# Train a DecisionTree model.
dt = DecisionTreeRegressor(featuresCol="indexedFeatures")

# Chain indexer and tree in a Pipeline
pipeline = Pipeline(stages=[featureIndexer, dt])

# Train model.  This also runs the indexer.
model = pipeline.fit(trainingData)

# Make predictions.
predictions = model.transform(testData)

# Select example rows to display.
predictions.select("prediction", "label", "features").show(5)

# Select (prediction, true label) and compute test error
evaluator = RegressionEvaluator(
    labelCol="label", predictionCol="prediction", metricName="rmse")
rmse = evaluator.evaluate(predictions)
print("Root Mean Squared Error (RMSE) on test data = %g" % rmse)

treeModel = model.stages[1]
# summary only
print(treeModel)

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/decision_tree_regression_example.py"에서 찾을 수 있어요.

파라미터에 대한 자세한 내용은 Scala API 문서에서 찾을 수 있어요.

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.feature.VectorIndexer
import org.apache.spark.ml.regression.DecisionTreeRegressionModel
import org.apache.spark.ml.regression.DecisionTreeRegressor

// Load the data stored in LIBSVM format as a DataFrame.
val data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

// Automatically identify categorical features, and index them.
// Here, we treat features with > 4 distinct values as continuous.
val featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4)
  .fit(data)

// Split the data into training and test sets (30% held out for testing).
val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))

// Train a DecisionTree model.
val dt = new DecisionTreeRegressor()
  .setLabelCol("label")
  .setFeaturesCol("indexedFeatures")

// Chain indexer and tree in a Pipeline.
val pipeline = new Pipeline()
  .setStages(Array(featureIndexer, dt))

// Train model. This also runs the indexer.
val model = pipeline.fit(trainingData)

// Make predictions.
val predictions = model.transform(testData)

// Select example rows to display.
predictions.select("prediction", "label", "features").show(5)

// Select (prediction, true label) and compute test error.
val evaluator = new RegressionEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("rmse")
val rmse = evaluator.evaluate(predictions)
println(s"Root Mean Squared Error (RMSE) on test data = $rmse")

val treeModel = model.stages(1).asInstanceOf[DecisionTreeRegressionModel]
println(s"Learned regression tree model:\n ${treeModel.toDebugString}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/DecisionTreeRegressionExample.scala"에서 찾을 수 있어요.

파라미터에 대한 자세한 내용은 Java API 문서에서 찾을 수 있어요.

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.evaluation.RegressionEvaluator;
import org.apache.spark.ml.feature.VectorIndexer;
import org.apache.spark.ml.feature.VectorIndexerModel;
import org.apache.spark.ml.regression.DecisionTreeRegressionModel;
import org.apache.spark.ml.regression.DecisionTreeRegressor;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load the data stored in LIBSVM format as a DataFrame.
Dataset<Row> data = spark.read().format("libsvm")
  .load("data/mllib/sample_libsvm_data.txt");

// Automatically identify categorical features, and index them.
// Set maxCategories so features with > 4 distinct values are treated as continuous.
VectorIndexerModel featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4)
  .fit(data);

// Split the data into training and test sets (30% held out for testing).
Dataset<Row>[] splits = data.randomSplit(new double[]{0.7, 0.3});
Dataset<Row> trainingData = splits[0];
Dataset<Row> testData = splits[1];

// Train a DecisionTree model.
DecisionTreeRegressor dt = new DecisionTreeRegressor()
  .setFeaturesCol("indexedFeatures");

// Chain indexer and tree in a Pipeline.
Pipeline pipeline = new Pipeline()
  .setStages(new PipelineStage[]{featureIndexer, dt});

// Train model. This also runs the indexer.
PipelineModel model = pipeline.fit(trainingData);

// Make predictions.
Dataset<Row> predictions = model.transform(testData);

// Select example rows to display.
predictions.select("label", "features").show(5);

// Select (prediction, true label) and compute test error.
RegressionEvaluator evaluator = new RegressionEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("rmse");
double rmse = evaluator.evaluate(predictions);
System.out.println("Root Mean Squared Error (RMSE) on test data = " + rmse);

DecisionTreeRegressionModel treeModel =
  (DecisionTreeRegressionModel) (model.stages()[1]);
System.out.println("Learned regression tree model:\n" + treeModel.toDebugString());

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaDecisionTreeRegressionExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

# Load training data
df <- read.df("data/mllib/sample_linear_regression_data.txt", source = "libsvm")
training <- df
test <- df

# Fit a DecisionTree regression model with spark.decisionTree
model <- spark.decisionTree(training, label ~ features, "regression")

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/decisionTree.R"에서 찾을 수 있어요.

랜덤 포레스트 회귀(Random forest regression)

랜덤 포레스트는 인기 있는 분류·회귀 메서드 패밀리예요. spark.ml 구현에 대한 더 많은 정보는 랜덤 포레스트 섹션에서 찾을 수 있어요.

예제

다음 예제는 LibSVM 형식의 데이터셋을 로드하고, 훈련·테스트 집합으로 나누고, 첫 데이터셋에서 훈련한 뒤 분리된 테스트 집합에서 평가해요. 트리 기반 알고리즘이 인식할 수 있는 메타데이터를 DataFrame에 추가하는 피처 변환기를 사용해요.

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml import Pipeline
from pyspark.ml.regression import RandomForestRegressor
from pyspark.ml.feature import VectorIndexer
from pyspark.ml.evaluation import RegressionEvaluator

# Load and parse the data file, converting it to a DataFrame.
data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

# Automatically identify categorical features, and index them.
# Set maxCategories so features with > 4 distinct values are treated as continuous.
featureIndexer =\
    VectorIndexer(inputCol="features", outputCol="indexedFeatures", maxCategories=4).fit(data)

# Split the data into training and test sets (30% held out for testing)
(trainingData, testData) = data.randomSplit([0.7, 0.3])

# Train a RandomForest model.
rf = RandomForestRegressor(featuresCol="indexedFeatures")

# Chain indexer and forest in a Pipeline
pipeline = Pipeline(stages=[featureIndexer, rf])

# Train model.  This also runs the indexer.
model = pipeline.fit(trainingData)

# Make predictions.
predictions = model.transform(testData)

# Select example rows to display.
predictions.select("prediction", "label", "features").show(5)

# Select (prediction, true label) and compute test error
evaluator = RegressionEvaluator(
    labelCol="label", predictionCol="prediction", metricName="rmse")
rmse = evaluator.evaluate(predictions)
print("Root Mean Squared Error (RMSE) on test data = %g" % rmse)

rfModel = model.stages[1]
print(rfModel)  # summary only

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/random_forest_regressor_example.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.feature.VectorIndexer
import org.apache.spark.ml.regression.{RandomForestRegressionModel, RandomForestRegressor}

// Load and parse the data file, converting it to a DataFrame.
val data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

// Automatically identify categorical features, and index them.
// Set maxCategories so features with > 4 distinct values are treated as continuous.
val featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4)
  .fit(data)

// Split the data into training and test sets (30% held out for testing).
val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))

// Train a RandomForest model.
val rf = new RandomForestRegressor()
  .setLabelCol("label")
  .setFeaturesCol("indexedFeatures")

// Chain indexer and forest in a Pipeline.
val pipeline = new Pipeline()
  .setStages(Array(featureIndexer, rf))

// Train model. This also runs the indexer.
val model = pipeline.fit(trainingData)

// Make predictions.
val predictions = model.transform(testData)

// Select example rows to display.
predictions.select("prediction", "label", "features").show(5)

// Select (prediction, true label) and compute test error.
val evaluator = new RegressionEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("rmse")
val rmse = evaluator.evaluate(predictions)
println(s"Root Mean Squared Error (RMSE) on test data = $rmse")

val rfModel = model.stages(1).asInstanceOf[RandomForestRegressionModel]
println(s"Learned regression forest model:\n ${rfModel.toDebugString}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/RandomForestRegressorExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.evaluation.RegressionEvaluator;
import org.apache.spark.ml.feature.VectorIndexer;
import org.apache.spark.ml.feature.VectorIndexerModel;
import org.apache.spark.ml.regression.RandomForestRegressionModel;
import org.apache.spark.ml.regression.RandomForestRegressor;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load and parse the data file, converting it to a DataFrame.
Dataset<Row> data = spark.read().format("libsvm").load("data/mllib/sample_libsvm_data.txt");

// Automatically identify categorical features, and index them.
// Set maxCategories so features with > 4 distinct values are treated as continuous.
VectorIndexerModel featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4)
  .fit(data);

// Split the data into training and test sets (30% held out for testing)
Dataset<Row>[] splits = data.randomSplit(new double[] {0.7, 0.3});
Dataset<Row> trainingData = splits[0];
Dataset<Row> testData = splits[1];

// Train a RandomForest model.
RandomForestRegressor rf = new RandomForestRegressor()
  .setLabelCol("label")
  .setFeaturesCol("indexedFeatures");

// Chain indexer and forest in a Pipeline
Pipeline pipeline = new Pipeline()
  .setStages(new PipelineStage[] {featureIndexer, rf});

// Train model. This also runs the indexer.
PipelineModel model = pipeline.fit(trainingData);

// Make predictions.
Dataset<Row> predictions = model.transform(testData);

// Select example rows to display.
predictions.select("prediction", "label", "features").show(5);

// Select (prediction, true label) and compute test error
RegressionEvaluator evaluator = new RegressionEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("rmse");
double rmse = evaluator.evaluate(predictions);
System.out.println("Root Mean Squared Error (RMSE) on test data = " + rmse);

RandomForestRegressionModel rfModel = (RandomForestRegressionModel)(model.stages()[1]);
System.out.println("Learned regression forest model:\n" + rfModel.toDebugString());

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaRandomForestRegressorExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

# Load training data
df <- read.df("data/mllib/sample_linear_regression_data.txt", source = "libsvm")
training <- df
test <- df

# Fit a random forest regression model with spark.randomForest
model <- spark.randomForest(training, label ~ features, "regression", numTrees = 10)

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/randomForest.R"에서 찾을 수 있어요.

그래디언트 부스티드 트리 회귀(Gradient-boosted tree regression)

그래디언트 부스티드 트리(GBT)는 의사결정 트리 앙상블을 사용하는 인기 있는 회귀 메서드예요. spark.ml 구현에 대한 더 많은 정보는 GBT 섹션에서 찾을 수 있어요.

예제

참고: 이 예제 데이터셋에 대해 GBTRegressor는 실제로 1회 반복만 필요하지만, 일반적으로는 그렇지 않아요.

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml import Pipeline
from pyspark.ml.regression import GBTRegressor
from pyspark.ml.feature import VectorIndexer
from pyspark.ml.evaluation import RegressionEvaluator

# Load and parse the data file, converting it to a DataFrame.
data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

# Automatically identify categorical features, and index them.
# Set maxCategories so features with > 4 distinct values are treated as continuous.
featureIndexer =\
    VectorIndexer(inputCol="features", outputCol="indexedFeatures", maxCategories=4).fit(data)

# Split the data into training and test sets (30% held out for testing)
(trainingData, testData) = data.randomSplit([0.7, 0.3])

# Train a GBT model.
gbt = GBTRegressor(featuresCol="indexedFeatures", maxIter=10)

# Chain indexer and GBT in a Pipeline
pipeline = Pipeline(stages=[featureIndexer, gbt])

# Train model.  This also runs the indexer.
model = pipeline.fit(trainingData)

# Make predictions.
predictions = model.transform(testData)

# Select example rows to display.
predictions.select("prediction", "label", "features").show(5)

# Select (prediction, true label) and compute test error
evaluator = RegressionEvaluator(
    labelCol="label", predictionCol="prediction", metricName="rmse")
rmse = evaluator.evaluate(predictions)
print("Root Mean Squared Error (RMSE) on test data = %g" % rmse)

gbtModel = model.stages[1]
print(gbtModel)  # summary only

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/gradient_boosted_tree_regressor_example.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.feature.VectorIndexer
import org.apache.spark.ml.regression.{GBTRegressionModel, GBTRegressor}

// Load and parse the data file, converting it to a DataFrame.
val data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

// Automatically identify categorical features, and index them.
// Set maxCategories so features with > 4 distinct values are treated as continuous.
val featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4)
  .fit(data)

// Split the data into training and test sets (30% held out for testing).
val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))

// Train a GBT model.
val gbt = new GBTRegressor()
  .setLabelCol("label")
  .setFeaturesCol("indexedFeatures")
  .setMaxIter(10)

// Chain indexer and GBT in a Pipeline.
val pipeline = new Pipeline()
  .setStages(Array(featureIndexer, gbt))

// Train model. This also runs the indexer.
val model = pipeline.fit(trainingData)

// Make predictions.
val predictions = model.transform(testData)

// Select example rows to display.
predictions.select("prediction", "label", "features").show(5)

// Select (prediction, true label) and compute test error.
val evaluator = new RegressionEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("rmse")
val rmse = evaluator.evaluate(predictions)
println(s"Root Mean Squared Error (RMSE) on test data = $rmse")

val gbtModel = model.stages(1).asInstanceOf[GBTRegressionModel]
println(s"Learned regression GBT model:\n ${gbtModel.toDebugString}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/GradientBoostedTreeRegressorExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.evaluation.RegressionEvaluator;
import org.apache.spark.ml.feature.VectorIndexer;
import org.apache.spark.ml.feature.VectorIndexerModel;
import org.apache.spark.ml.regression.GBTRegressionModel;
import org.apache.spark.ml.regression.GBTRegressor;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load and parse the data file, converting it to a DataFrame.
Dataset<Row> data = spark.read().format("libsvm").load("data/mllib/sample_libsvm_data.txt");

// Automatically identify categorical features, and index them.
// Set maxCategories so features with > 4 distinct values are treated as continuous.
VectorIndexerModel featureIndexer = new VectorIndexer()
  .setInputCol("features")
  .setOutputCol("indexedFeatures")
  .setMaxCategories(4)
  .fit(data);

// Split the data into training and test sets (30% held out for testing).
Dataset<Row>[] splits = data.randomSplit(new double[] {0.7, 0.3});
Dataset<Row> trainingData = splits[0];
Dataset<Row> testData = splits[1];

// Train a GBT model.
GBTRegressor gbt = new GBTRegressor()
  .setLabelCol("label")
  .setFeaturesCol("indexedFeatures")
  .setMaxIter(10);

// Chain indexer and GBT in a Pipeline.
Pipeline pipeline = new Pipeline().setStages(new PipelineStage[] {featureIndexer, gbt});

// Train model. This also runs the indexer.
PipelineModel model = pipeline.fit(trainingData);

// Make predictions.
Dataset<Row> predictions = model.transform(testData);

// Select example rows to display.
predictions.select("prediction", "label", "features").show(5);

// Select (prediction, true label) and compute test error.
RegressionEvaluator evaluator = new RegressionEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("rmse");
double rmse = evaluator.evaluate(predictions);
System.out.println("Root Mean Squared Error (RMSE) on test data = " + rmse);

GBTRegressionModel gbtModel = (GBTRegressionModel)(model.stages()[1]);
System.out.println("Learned regression GBT model:\n" + gbtModel.toDebugString());

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaGradientBoostedTreeRegressorExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

# Load training data
df <- read.df("data/mllib/sample_linear_regression_data.txt", source = "libsvm")
training <- df
test <- df

# Fit a GBT regression model with spark.gbt
model <- spark.gbt(training, label ~ features, "regression", maxIter = 10)

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/gbt.R"에서 찾을 수 있어요.

생존 회귀(Survival regression)

spark.ml에서 우리는 검열된 데이터(censored data)를 위한 모수적 생존 회귀 모델인 가속 실패 시간(Accelerated failure time, AFT) 모델을 구현해요. 이 모델은 생존 시간의 로그를 모델링하므로, 생존 분석을 위한 log-linear 모델이라고 자주 불려요. 같은 목적으로 설계된 Proportional hazards 모델과 달리 AFT 모델은 각 인스턴스가 목적 함수에 독립적으로 기여하므로 병렬화하기 더 쉬워요.

공변량 $x^{'}$의 값이 주어지고, 오른쪽 검열이 가능한 피험자 i = 1, …, n의 랜덤 수명 $t_{i}$에 대해 AFT 모델 하의 가능도 함수는 다음과 같아요.

\[ L(\beta,\sigma)=\prod_{i=1}^n[\frac{1}{\sigma}f_{0}(\frac{\log{t_{i}}-x^{'}\beta}{\sigma})]^{\delta_{i}}S_{0}(\frac{\log{t_{i}}-x^{'}\beta}{\sigma})^{1-\delta_{i}} \]

여기서 $\delta_{i}$는 사건이 발생했는지(즉, 검열되지 않았는지)의 표시자예요. $\epsilon_{i}=\frac{\log{t_{i}}-x^{'}\beta}{\sigma}$를 사용하면 로그 가능도 함수는 다음 형태를 취해요.

\[ \iota(\beta,\sigma)=\sum_{i=1}^{n}[-\delta_{i}\log\sigma+\delta_{i}\log{f_{0}}(\epsilon_{i})+(1-\delta_{i})\log{S_{0}(\epsilon_{i})}] \]

여기서 $S_{0}(\epsilon_{i})$는 기준 생존 함수이고, $f_{0}(\epsilon_{i})$는 해당 밀도 함수예요.

가장 흔히 사용되는 AFT 모델은 생존 시간의 Weibull 분포에 기반해요. 수명의 Weibull 분포는 수명 로그의 극단값 분포에 해당하며, $S_{0}(\epsilon)$ 함수는 다음과 같아요.

\[ S_{0}(\epsilon_{i})=\exp(-e^{\epsilon_{i}}) \]

$f_{0}(\epsilon_{i})$ 함수는 다음과 같아요.

\[ f_{0}(\epsilon_{i})=e^{\epsilon_{i}}\exp(-e^{\epsilon_{i}}) \]

Weibull 수명 분포를 갖는 AFT 모델의 로그 가능도 함수는 다음과 같아요.

\[ \iota(\beta,\sigma)= -\sum_{i=1}^n[\delta_{i}\log\sigma-\delta_{i}\epsilon_{i}+e^{\epsilon_{i}}] \]

음의 로그 가능도를 최소화하는 것이 최대 사후 확률과 동등하므로, 최적화에 사용하는 손실 함수는 $-\iota(\beta,\sigma)$예요. $\beta$와 $\log\sigma$에 대한 그라디언트 함수는 각각 다음과 같아요.

\[ \frac{\partial (-\iota)}{\partial \beta}=\sum_{1=1}^{n}[\delta_{i}-e^{\epsilon_{i}}]\frac{x_{i}}{\sigma} \]

\[ \frac{\partial (-\iota)}{\partial (\log\sigma)}=\sum_{i=1}^{n}[\delta_{i}+(\delta_{i}-e^{\epsilon_{i}})\epsilon_{i}] \]

AFT 모델은 볼록 최적화 문제로 공식화될 수 있어요. 즉, 계수 벡터 $\beta$와 로그 스케일 파라미터 $\log\sigma$에 의존하는 볼록 함수 $-\iota(\beta,\sigma)$의 최소화자 찾기예요. 구현의 기본 최적화 알고리즘은 L-BFGS예요. 구현은 R의 survival 함수 survreg와 결과가 일치해요.

절편 없이 상수가 아닌 0이 아닌 컬럼이 있는 데이터셋에서 AFTSurvivalRegressionModel을 피팅할 때, Spark MLlib는 상수 0이 아닌 컬럼에 대해 0 계수를 출력해요. 이 동작은 R survival::survreg와 달라요.

예제

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml.regression import AFTSurvivalRegression
from pyspark.ml.linalg import Vectors

training = spark.createDataFrame([
    (1.218, 1.0, Vectors.dense(1.560, -0.605)),
    (2.949, 0.0, Vectors.dense(0.346, 2.158)),
    (3.627, 0.0, Vectors.dense(1.380, 0.231)),
    (0.273, 1.0, Vectors.dense(0.520, 1.151)),
    (4.199, 0.0, Vectors.dense(0.795, -0.226))], ["label", "censor", "features"])
quantileProbabilities = [0.3, 0.6]
aft = AFTSurvivalRegression(quantileProbabilities=quantileProbabilities,
                            quantilesCol="quantiles")

model = aft.fit(training)

# Print the coefficients, intercept and scale parameter for AFT survival regression
print("Coefficients: " + str(model.coefficients))
print("Intercept: " + str(model.intercept))
print("Scale: " + str(model.scale))
model.transform(training).show(truncate=False)

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/aft_survival_regression.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.regression.AFTSurvivalRegression

val training = spark.createDataFrame(Seq(
  (1.218, 1.0, Vectors.dense(1.560, -0.605)),
  (2.949, 0.0, Vectors.dense(0.346, 2.158)),
  (3.627, 0.0, Vectors.dense(1.380, 0.231)),
  (0.273, 1.0, Vectors.dense(0.520, 1.151)),
  (4.199, 0.0, Vectors.dense(0.795, -0.226))
)).toDF("label", "censor", "features")
val quantileProbabilities = Array(0.3, 0.6)
val aft = new AFTSurvivalRegression()
  .setQuantileProbabilities(quantileProbabilities)
  .setQuantilesCol("quantiles")

val model = aft.fit(training)

// Print the coefficients, intercept and scale parameter for AFT survival regression
println(s"Coefficients: ${model.coefficients}")
println(s"Intercept: ${model.intercept}")
println(s"Scale: ${model.scale}")
model.transform(training).show(false)

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/AFTSurvivalRegressionExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import java.util.Arrays;
import java.util.List;

import org.apache.spark.ml.regression.AFTSurvivalRegression;
import org.apache.spark.ml.regression.AFTSurvivalRegressionModel;
import org.apache.spark.ml.linalg.VectorUDT;
import org.apache.spark.ml.linalg.Vectors;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
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> data = Arrays.asList(
  RowFactory.create(1.218, 1.0, Vectors.dense(1.560, -0.605)),
  RowFactory.create(2.949, 0.0, Vectors.dense(0.346, 2.158)),
  RowFactory.create(3.627, 0.0, Vectors.dense(1.380, 0.231)),
  RowFactory.create(0.273, 1.0, Vectors.dense(0.520, 1.151)),
  RowFactory.create(4.199, 0.0, Vectors.dense(0.795, -0.226))
);
StructType schema = new StructType(new StructField[]{
  new StructField("label", DataTypes.DoubleType, false, Metadata.empty()),
  new StructField("censor", DataTypes.DoubleType, false, Metadata.empty()),
  new StructField("features", new VectorUDT(), false, Metadata.empty())
});
Dataset<Row> training = spark.createDataFrame(data, schema);
double[] quantileProbabilities = new double[]{0.3, 0.6};
AFTSurvivalRegression aft = new AFTSurvivalRegression()
  .setQuantileProbabilities(quantileProbabilities)
  .setQuantilesCol("quantiles");

AFTSurvivalRegressionModel model = aft.fit(training);

// Print the coefficients, intercept and scale parameter for AFT survival regression
System.out.println("Coefficients: " + model.coefficients());
System.out.println("Intercept: " + model.intercept());
System.out.println("Scale: " + model.scale());
model.transform(training).show(false);

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaAFTSurvivalRegressionExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

# Use the ovarian dataset available in R survival package
library(survival)

# Fit an accelerated failure time (AFT) survival regression model with spark.survreg
ovarianDF <- suppressWarnings(createDataFrame(ovarian))
aftDF <- ovarianDF
aftTestDF <- ovarianDF
aftModel <- spark.survreg(aftDF, Surv(futime, fustat) ~ ecog_ps + rx)

# Model summary
summary(aftModel)

# Prediction
aftPredictions <- predict(aftModel, aftTestDF)
head(aftPredictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/survreg.R"에서 찾을 수 있어요.

단조 회귀(Isotonic regression)

단조 회귀는 회귀 알고리즘 패밀리에 속해요. 공식적으로 단조 회귀는 관측된 응답을 나타내는 유한 실수 집합 $Y = {y_1, y_2, ..., y_n}$과 피팅할 미지의 응답 값 $X = {x_1, x_2, ..., x_n}$이 주어졌을 때, 다음을 최소화하는 함수를 찾는 문제예요.

\begin{equation} f(x) = \sum_{i=1}^n w_i (y_i - x_i)^2 \end{equation}

$x_1\le x_2\le ...\le x_n$이라는 완전 순서 제약 하에서 말이죠. 여기서 $w_i$는 양의 가중치예요. 결과 함수를 단조 회귀라고 하며 독특해요. 이는 순서 제약 하의 최소 제곱 문제로 볼 수 있어요. 본질적으로 단조 회귀는 원래 데이터 포인트에 가장 잘 맞는 단조 함수예요.

우리는 pool adjacent violators 알고리즘을 구현하는데, 이는 단조 회귀 병렬화 접근법을 사용해요. 훈련 입력은 label, features, weight 세 컬럼을 포함하는 DataFrame이에요. 또한 IsotonicRegression 알고리즘에는 기본이 true인 $isotonic$이라는 선택적 파라미터가 하나 있어요. 이 인자는 단조 회귀가 isotonic(단조 증가)인지 antitonic(단조 감소)인지 지정해요.

훈련은 알려진·미지의 피처 모두에 대한 레이블 예측에 쓸 수 있는 IsotonicRegressionModel을 반환해요. 단조 회귀의 결과는 조각별 선형 함수로 처리돼요. 따라서 예측 규칙은 다음과 같아요.

• 예측 입력이 훈련 피처와 정확히 일치하면 관련 예측이 반환돼요. 같은 피처에 여러 예측이 있으면 그 중 하나가 반환돼요. 어느 것이 반환될지는 정의되지 않아요(java.util.Arrays.binarySearch와 동일).
• 예측 입력이 모든 훈련 피처보다 낮거나 높으면 각각 가장 낮거나 가장 높은 피처의 예측이 반환돼요. 같은 피처에 여러 예측이 있으면 각각 가장 낮거나 가장 높은 값이 반환돼요.
• 예측 입력이 두 훈련 피처 사이에 있으면 예측이 조각별 선형 함수로 처리되고 가장 가까운 두 피처의 예측에서 보간된 값이 계산돼요. 같은 피처에 여러 값이 있으면 이전 시점과 같은 규칙이 사용돼요.

예제

API에 대한 자세한 내용은 IsotonicRegression Python 문서를 참고하세요.

from pyspark.ml.regression import IsotonicRegression

# Loads data.
dataset = spark.read.format("libsvm")\
    .load("data/mllib/sample_isotonic_regression_libsvm_data.txt")

# Trains an isotonic regression model.
model = IsotonicRegression().fit(dataset)
print("Boundaries in increasing order: %s\n" % str(model.boundaries))
print("Predictions associated with the boundaries: %s\n" % str(model.predictions))

# Makes predictions.
model.transform(dataset).show()

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/isotonic_regression_example.py"에서 찾을 수 있어요.

API에 대한 자세한 내용은 IsotonicRegression Scala 문서를 참고하세요.

import org.apache.spark.ml.regression.IsotonicRegression

// Loads data.
val dataset = spark.read.format("libsvm")
  .load("data/mllib/sample_isotonic_regression_libsvm_data.txt")

// Trains an isotonic regression model.
val ir = new IsotonicRegression()
val model = ir.fit(dataset)

println(s"Boundaries in increasing order: ${model.boundaries}\n")
println(s"Predictions associated with the boundaries: ${model.predictions}\n")

// Makes predictions.
model.transform(dataset).show()

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/IsotonicRegressionExample.scala"에서 찾을 수 있어요.

API에 대한 자세한 내용은 IsotonicRegression Java 문서를 참고하세요.

import org.apache.spark.ml.regression.IsotonicRegression;
import org.apache.spark.ml.regression.IsotonicRegressionModel;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

// Loads data.
Dataset<Row> dataset = spark.read().format("libsvm")
  .load("data/mllib/sample_isotonic_regression_libsvm_data.txt");

// Trains an isotonic regression model.
IsotonicRegression ir = new IsotonicRegression();
IsotonicRegressionModel model = ir.fit(dataset);

System.out.println("Boundaries in increasing order: " + model.boundaries() + "\n");
System.out.println("Predictions associated with the boundaries: " + model.predictions() + "\n");

// Makes predictions.
model.transform(dataset).show();

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaIsotonicRegressionExample.java"에서 찾을 수 있어요.

API에 대한 자세한 내용은 IsotonicRegression R API 문서를 참고하세요.

# Load training data
df <- read.df("data/mllib/sample_isotonic_regression_libsvm_data.txt", source = "libsvm")
training <- df
test <- df

# Fit an isotonic regression model with spark.isoreg
model <- spark.isoreg(training, label ~ features, isotonic = FALSE)

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/isoreg.R"에서 찾을 수 있어요.

팩터라이제이션 머신 회귀(Factorization machines regressor)

팩터라이제이션 머신 구현에 대한 더 많은 배경과 세부 사항은 Factorization Machines 섹션을 참고하세요.

예제

다음 예제는 LibSVM 형식의 데이터셋을 로드하고, 훈련·테스트 집합으로 나누고, 첫 데이터셋에서 훈련한 뒤 분리된 테스트 집합에서 평가해요. 폭주하는 그라디언트 문제를 막기 위해 피처를 0과 1 사이로 스케일링해요.

자세한 내용은 Python API 문서를 참고하세요.

from pyspark.ml import Pipeline
from pyspark.ml.regression import FMRegressor
from pyspark.ml.feature import MinMaxScaler
from pyspark.ml.evaluation import RegressionEvaluator

# Load and parse the data file, converting it to a DataFrame.
data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

# Scale features.
featureScaler = MinMaxScaler(inputCol="features", outputCol="scaledFeatures").fit(data)

# Split the data into training and test sets (30% held out for testing)
(trainingData, testData) = data.randomSplit([0.7, 0.3])

# Train a FM model.
fm = FMRegressor(featuresCol="scaledFeatures", stepSize=0.001)

# Create a Pipeline.
pipeline = Pipeline(stages=[featureScaler, fm])

# Train model.
model = pipeline.fit(trainingData)

# Make predictions.
predictions = model.transform(testData)

# Select example rows to display.
predictions.select("prediction", "label", "features").show(5)

# Select (prediction, true label) and compute test error
evaluator = RegressionEvaluator(
    labelCol="label", predictionCol="prediction", metricName="rmse")
rmse = evaluator.evaluate(predictions)
print("Root Mean Squared Error (RMSE) on test data = %g" % rmse)

fmModel = model.stages[1]
print("Factors: " + str(fmModel.factors))  # type: ignore
print("Linear: " + str(fmModel.linear))  # type: ignore
print("Intercept: " + str(fmModel.intercept))  # type: ignore

전체 예제 코드는 Spark 저장소의 "examples/src/main/python/ml/fm_regressor_example.py"에서 찾을 수 있어요.

자세한 내용은 Scala API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.feature.MinMaxScaler
import org.apache.spark.ml.regression.{FMRegressionModel, FMRegressor}

// Load and parse the data file, converting it to a DataFrame.
val data = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")

// Scale features.
val featureScaler = new MinMaxScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")
  .fit(data)

// Split the data into training and test sets (30% held out for testing).
val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))

// Train a FM model.
val fm = new FMRegressor()
  .setLabelCol("label")
  .setFeaturesCol("scaledFeatures")
  .setStepSize(0.001)

// Create a Pipeline.
val pipeline = new Pipeline()
  .setStages(Array(featureScaler, fm))

// Train model.
val model = pipeline.fit(trainingData)

// Make predictions.
val predictions = model.transform(testData)

// Select example rows to display.
predictions.select("prediction", "label", "features").show(5)

// Select (prediction, true label) and compute test error.
val evaluator = new RegressionEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("rmse")
val rmse = evaluator.evaluate(predictions)
println(s"Root Mean Squared Error (RMSE) on test data = $rmse")

val fmModel = model.stages(1).asInstanceOf[FMRegressionModel]
println(s"Factors: ${fmModel.factors} Linear: ${fmModel.linear} " +
  s"Intercept: ${fmModel.intercept}")

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/ml/FMRegressorExample.scala"에서 찾을 수 있어요.

자세한 내용은 Java API 문서를 참고하세요.

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.evaluation.RegressionEvaluator;
import org.apache.spark.ml.feature.MinMaxScaler;
import org.apache.spark.ml.feature.MinMaxScalerModel;
import org.apache.spark.ml.regression.FMRegressionModel;
import org.apache.spark.ml.regression.FMRegressor;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// Load and parse the data file, converting it to a DataFrame.
Dataset<Row> data = spark.read().format("libsvm").load("data/mllib/sample_libsvm_data.txt");

// Scale features.
MinMaxScalerModel featureScaler = new MinMaxScaler()
    .setInputCol("features")
    .setOutputCol("scaledFeatures")
    .fit(data);

// Split the data into training and test sets (30% held out for testing).
Dataset<Row>[] splits = data.randomSplit(new double[] {0.7, 0.3});
Dataset<Row> trainingData = splits[0];
Dataset<Row> testData = splits[1];

// Train a FM model.
FMRegressor fm = new FMRegressor()
    .setLabelCol("label")
    .setFeaturesCol("scaledFeatures")
    .setStepSize(0.001);

// Create a Pipeline.
Pipeline pipeline = new Pipeline().setStages(new PipelineStage[] {featureScaler, fm});

// Train model.
PipelineModel model = pipeline.fit(trainingData);

// Make predictions.
Dataset<Row> predictions = model.transform(testData);

// Select example rows to display.
predictions.select("prediction", "label", "features").show(5);

// Select (prediction, true label) and compute test error.
RegressionEvaluator evaluator = new RegressionEvaluator()
    .setLabelCol("label")
    .setPredictionCol("prediction")
    .setMetricName("rmse");
double rmse = evaluator.evaluate(predictions);
System.out.println("Root Mean Squared Error (RMSE) on test data = " + rmse);

FMRegressionModel fmModel = (FMRegressionModel)(model.stages()[1]);
System.out.println("Factors: " + fmModel.factors());
System.out.println("Linear: " + fmModel.linear());
System.out.println("Intercept: " + fmModel.intercept());

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/ml/JavaFMRegressorExample.java"에서 찾을 수 있어요.

자세한 내용은 R API 문서를 참고하세요.

참고: 현재 SparkR은 피처 스케일링을 지원하지 않아요.

# Load training data
df <- read.df("data/mllib/sample_linear_regression_data.txt", source = "libsvm")
training_test <- randomSplit(df, c(0.7, 0.3))
training <- training_test[[1]]
test <- training_test[[2]]

# Fit a FM regression model
model <- spark.fmRegressor(training, label ~ features)

# Model summary
summary(model)

# Prediction
predictions <- predict(model, test)
head(predictions)

전체 예제 코드는 Spark 저장소의 "examples/src/main/r/ml/fmRegressor.R"에서 찾을 수 있어요.

선형 메서드(Linear methods)

우리는 $L_1$ 또는 $L_2$ 정규화를 가진 로지스틱 회귀와 선형 최소 제곱 같은 인기 있는 선형 메서드를 구현해요. 구현과 튜닝에 대한 자세한 내용은 RDD 기반 API용 선형 메서드 가이드를 참고하세요. 이 정보는 여전히 관련돼요.

우리는 또한 Zou 등이 제안한 $L_1$과 $L_2$ 정규화의 혼합인 Elastic net에 대한 DataFrame API를 포함해요. 수학적으로 $L_1$과 $L_2$ 정규화 항의 볼록 결합으로 정의돼요.

\[ \alpha \left( \lambda \|\wv\|_1 \right) + (1-\alpha) \left( \frac{\lambda}{2}\|\wv\|_2^2 \right) , \alpha \in [0, 1], \lambda \geq 0 \]

$\alpha$를 적절히 설정하면 elastic net은 $L_1$과 $L_2$ 정규화를 특수한 경우로 모두 포함해요. 예를 들어 선형 회귀 모델을 elastic net 파라미터 $\alpha$를 1로 설정해 훈련하면 Lasso 모델과 동등해요. 반면 $\alpha$를 0으로 설정하면 훈련된 모델은 ridge 회귀 모델로 줄어들어요. 우리는 elastic net 정규화로 선형 회귀와 로지스틱 회귀 모두에 Pipelines API를 구현해요.

팩터라이제이션 머신(Factorization Machines)

팩터라이제이션 머신은 엄청난 희소성(광고, 추천 시스템 같은)이 있는 문제에서도 피처 간 상호작용을 추정할 수 있어요. spark.ml 구현은 이진 분류와 회귀를 위한 팩터라이제이션 머신을 지원해요.

팩터라이제이션 머신 공식은 다음과 같아요.

\[\hat{y} = w_0 + \sum\limits^n_{i-1} w_i x_i + \sum\limits^n_{i=1} \sum\limits^n_{j=i+1} \langle v_i, v_j \rangle x_i x_j\]

처음 두 항은 절편과 선형 항(선형 회귀와 동일)을 나타내고, 마지막 항은 쌍별 상호작용 항을 나타내요. (v_i)는 k개의 팩터를 가진 i번째 변수를 설명해요.

FM은 회귀에 쓸 수 있으며 최적화 기준은 평균 제곱 오차예요. FM은 시그모이드 함수를 통해 이진 분류에도 쓸 수 있어요. 이때 최적화 기준은 로지스틱 손실이에요.

쌍별 상호작용은 재공식화될 수 있어요.

\[\sum\limits^n_{i=1} \sum\limits^n_{j=i+1} \langle v_i, v_j \rangle x_i x_j = \frac{1}{2}\sum\limits^k_{f=1} \left(\left( \sum\limits^n_{i=1}v_{i,f}x_i \right)^2 - \sum\limits^n_{i=1}v_{i,f}^2x_i^2 \right)\]

이 식은 k와 n 모두에서 선형 복잡도만 가져요. 즉 계산이 (O(kn))이에요.

일반적으로 폭주하는 그라디언트 문제를 막기 위해, 연속 피처를 0과 1 사이로 스케일링하거나 연속 피처를 구간화하고 원-핫 인코딩하는 것이 가장 좋아요.

의사결정 트리(Decision trees)

의사결정 트리와 그 앙상블은 분류·회귀 머신러닝 작업에 인기 있는 방법이에요. 의사결정 트리는 해석하기 쉽고, 범주형 피처를 다루고, 다중 클래스 분류 설정으로 확장되며, 피처 스케일링을 요구하지 않고, 비선형성과 피처 상호작용을 포착할 수 있어서 널리 사용돼요. 랜덤 포레스트와 부스팅 같은 트리 앙상블 알고리즘은 분류·회귀 작업에서 최고 성능자들 중 하나예요.

spark.ml 구현은 연속·범주형 피처를 모두 사용해 이진·다중 클래스 분류와 회귀를 위한 의사결정 트리를 지원해요. 구현은 데이터를 행별로 분할해 수백만 또는 수십억 개의 인스턴스로 분산 훈련을 가능하게 해요.

의사결정 트리 알고리즘에 대한 더 많은 정보는 MLlib Decision Tree 가이드에서 찾을 수 있어요. 이 API와 원래 MLlib Decision Tree API의 주요 차이점은 다음과 같아요.

• ML Pipelines 지원
• 분류용·회귀용 의사결정 트리의 분리
• 연속·범주형 피처 구분에 DataFrame 메타데이터 사용

의사결정 트리용 Pipelines API는 원래 API보다 약간 더 많은 기능을 제공해요. 특히 분류에서는 각 클래스의 예측 확률(일명 클래스 조건부 확률)을 얻을 수 있고, 회귀에서는 예측의 편향된 표본 분산을 얻을 수 있어요.

트리 앙상블(랜덤 포레스트와 그래디언트 부스티드 트리)은 아래 Tree ensembles 섹션에서 설명돼요.

입력과 출력(Inputs and Outputs)

여기서 입력·출력(예측) 컬럼 타입을 나열해요. 모든 출력 컬럼은 선택적이에요. 출력 컬럼을 제외하려면 해당 Param을 빈 문자열로 설정하세요.

입력 컬럼

Param 이름	타입	기본값	설명
labelCol	Double	"label"	예측할 레이블
featuresCol	Vector	"features"	피처 벡터

출력 컬럼

Param 이름	타입	기본값	설명	참고
predictionCol	Double	"prediction"	예측된 레이블
rawPredictionCol	Vector	"rawPrediction"	# 클래스 길이의 벡터. 예측을 만드는 트리 노드에서 훈련 인스턴스 레이블의 개수	분류만
probabilityCol	Vector	"probability"	multinomial 분포로 정규화된 rawPrediction과 같은 # 클래스 길이의 벡터	분류만
varianceCol	Double		예측의 편향된 표본 분산	회귀만

트리 앙상블(Tree Ensembles)

DataFrame API는 Random Forests와 Gradient-Boosted Trees (GBTs) 두 가지 주요 트리 앙상블 알고리즘을 지원해요. 둘 다 spark.ml 의사결정 트리를 기본 모델로 사용해요.

앙상블 알고리즘에 대한 더 많은 정보는 MLlib Ensemble 가이드에서 찾을 수 있어요. 이 섹션에서는 앙상블용 DataFrame API를 보여줘요.

이 API와 원래 MLlib ensembles API의 주요 차이점은 다음과 같아요.

• DataFrames와 ML Pipelines 지원
• 분류·회귀 분리
• 연속·범주형 피처 구분에 DataFrame 메타데이터 사용
• 랜덤 포레스트에 더 많은 기능: 피처 중요도 추정과 분류의 각 클래스 예측 확률(일명 클래스 조건부 확률)

랜덤 포레스트(Random Forests)

랜덤 포레스트는 의사결정 트리의 앙상블이에요. 랜덤 포레스트는 많은 의사결정 트리를 결합해 과적합 위험을 줄여요. spark.ml 구현은 연속·범주형 피처를 모두 사용해 이진·다중 클래스 분류와 회귀를 위한 랜덤 포레스트를 지원해요.

알고리즘 자체에 대한 더 많은 정보는 spark.mllib의 랜덤 포레스트 문서를 참고하세요.

입력과 출력

여기서 입력·출력(예측) 컬럼 타입을 나열해요. 모든 출력 컬럼은 선택적이에요. 출력 컬럼을 제외하려면 해당 Param을 빈 문자열로 설정하세요.

입력 컬럼

Param 이름	타입	기본값	설명
labelCol	Double	"label"	예측할 레이블
featuresCol	Vector	"features"	피처 벡터

출력 컬럼(예측)

Param 이름	타입	기본값	설명	참고
predictionCol	Double	"prediction"	예측된 레이블
rawPredictionCol	Vector	"rawPrediction"	# 클래스 길이의 벡터. 예측을 만드는 트리 노드에서 훈련 인스턴스 레이블의 개수	분류만
probabilityCol	Vector	"probability"	multinomial 분포로 정규화된 rawPrediction과 같은 # 클래스 길이의 벡터	분류만

그래디언트 부스티드 트리(GBTs)

그래디언트 부스티드 트리(GBTs)는 의사결정 트리의 앙상블이에요. GBT는 손실 함수를 최소화하기 위해 의사결정 트리를 반복적으로 훈련해요. spark.ml 구현은 연속·범주형 피처를 모두 사용해 이진 분류와 회귀를 위한 GBT를 지원해요.

알고리즘 자체에 대한 더 많은 정보는 spark.mllib의 GBT 문서를 참고하세요.

입력과 출력

여기서 입력·출력(예측) 컬럼 타입을 나열해요. 모든 출력 컬럼은 선택적이에요. 출력 컬럼을 제외하려면 해당 Param을 빈 문자열로 설정하세요.

입력 컬럼

Param 이름	타입	기본값	설명
labelCol	Double	"label"	예측할 레이블
featuresCol	Vector	"features"	피처 벡터

GBTClassifier는 현재 이진 레이블만 지원한다는 점에 유의하세요.

출력 컬럼(예측)

Param 이름	타입	기본값	설명	참고
predictionCol	Double	"prediction"	예측된 레이블

미래에 GBTClassifier는 RandomForestClassifier처럼 rawPrediction과 probability 컬럼도 출력할 거예요.

더 알아보기 (Learn more)

• ML 파이프라인 — 위 예제들이 쓰는 파이프라인 개념.
• MLlib Main Guide — MLlib의 전반적인 내용.
• ML Tuning Guide — 하이퍼파라미터 튜닝.