ML中的一個重要任務(wù)是模型選擇，或者使用數(shù)據(jù)為給定任務(wù)找到最佳的模型或參數(shù)。這也稱為調(diào)優(yōu)?？梢詫蝹€的估計器(如LogisticRegression?)進(jìn)行調(diào)優(yōu)，也可以對包括多種算法、特性化和其他步驟的整個pipeline?進(jìn)行調(diào)優(yōu)。用戶可以一次調(diào)優(yōu)整個Pipeline?，而不是分別調(diào)優(yōu) Pipeline 中的每個元素。

ML中的一個重要任務(wù)是模型選擇，或者使用數(shù)據(jù)為給定任務(wù)找到最佳的模型或參數(shù)。這也稱為調(diào)優(yōu)。可以對單個的Estimator?(如LogisticRegression?)進(jìn)行調(diào)優(yōu)，也可以對包括多種算法、特性化和其他步驟的整個pipeline?進(jìn)行調(diào)優(yōu)。用戶可以一次調(diào)優(yōu)整個Pipeline?，而不是分別調(diào)優(yōu)Pipeline中的每個元素。

MLlib支持使用CrossValidator和TrainValidationSplit等工具進(jìn)行模型選擇。這些工具需要具備以下條件:

• 估計器：要調(diào)優(yōu)的算法或管道pipeline
• 一組參數(shù)：可選擇的參數(shù)，有時稱為搜索的“參數(shù)網(wǎng)格”
• 評估者：度量擬合模型在測試數(shù)據(jù)上的表現(xiàn)

這些模型選擇工具的工作方式如下：

• 他們將輸入數(shù)據(jù)拆分為單獨(dú)的訓(xùn)練和測試數(shù)據(jù)集。
• 對于每個（訓(xùn)練、測試）對，它們遍歷ParamMap 集合：

對于每個ParamMap?，使用這些參數(shù)擬合Estimator?，得到擬合的Model?，并使用Evaluator? 評估Model的性能。

• 他們選擇Model由表現(xiàn)最好的一組參數(shù)產(chǎn)生。

為了幫助構(gòu)造參數(shù)網(wǎng)格，用戶可以使用ParamGridBuilder。默認(rèn)情況下，參數(shù)網(wǎng)格中的參數(shù)集以串行方式計算。在使用CrossValidator或TrainValidationSplit運(yùn)行模型選擇之前，可以通過將并行度設(shè)置為2或更多(1的值將是串行的)來并行地進(jìn)行參數(shù)評估。并行度的值應(yīng)該謹(jǐn)慎選擇，以便在不超過集群資源的情況下最大化并行度，較大的值不一定會提高性能。一般來說，10以上的值對大多數(shù)集群來說應(yīng)該足夠了。

交叉驗證

CrossValidator交叉驗證器首先將數(shù)據(jù)集分割為一組折疊數(shù)據(jù)集，這些折疊數(shù)據(jù)集用作單獨(dú)的訓(xùn)練數(shù)據(jù)集和測試數(shù)據(jù)集。例如，當(dāng)k=3次時，CrossValidator將生成3對(訓(xùn)練，測試)數(shù)據(jù)集，每對數(shù)據(jù)集使用2/3的數(shù)據(jù)進(jìn)行訓(xùn)練，1/3的數(shù)據(jù)進(jìn)行測試。為了評估一個特定的ParamMap, CrossValidator通過在3個不同的(訓(xùn)練，測試)數(shù)據(jù)集對上擬合Estimator產(chǎn)生的3個模型計算平均評估度量。

在確定最佳ParamMap之后，CrossValidator最終使用最佳ParamMap和整個數(shù)據(jù)集重新匹配Estimator。

from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.feature import HashingTF, Tokenizer
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder

# 準(zhǔn)備訓(xùn)練文件，并做好標(biāo)簽。
training = spark.createDataFrame([
    (0, "a b c d e spark", 1.0),
    (1, "b d", 0.0),
    (2, "spark f g h", 1.0),
    (3, "hadoop mapreduce", 0.0),
    (4, "b spark who", 1.0),
    (5, "g d a y", 0.0),
    (6, "spark fly", 1.0),
    (7, "was mapreduce", 0.0),
    (8, "e spark program", 1.0),
    (9, "a e c l", 0.0),
    (10, "spark compile", 1.0),
    (11, "hadoop software", 0.0)
], ["id", "text", "label"])

# 配置一個ML管道，它由樹stages組成:tokenizer、hashingTF和lr。
tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol="features")
lr = LogisticRegression(maxIter=10)
pipeline = Pipeline(stages=[tokenizer, hashingTF, lr])

# 我們現(xiàn)在將Pipeline作為一個Estimator，將其包裝在CrossValidator實例中。
# 這將允許我們共同選擇所有管道階段的參數(shù)。
# 交叉驗證器需要一個Estimator、一組Estimator ParamMaps和一個Evaluator。
# 我們使用ParamGridBuilder來構(gòu)造一個用于搜索的參數(shù)網(wǎng)格。
# hashingTF.numFeatures 的3個值, lr.regParam的2個值，
# 這個網(wǎng)格將有3 x 2 = 6的參數(shù)設(shè)置供CrossValidator選擇。

 
paramGrid = ParamGridBuilder() \
    .addGrid(hashingTF.numFeatures, [10, 100, 1000]) \
    .addGrid(lr.regParam, [0.1, 0.01]) \
    .build()

crossval = CrossValidator(estimator=pipeline,
                          estimatorParamMaps=paramGrid,
                          evaluator=BinaryClassificationEvaluator(),
                          numFolds=2)  # 使用3+ folds

# 運(yùn)行交叉驗證，并選擇最佳參數(shù)集。
cvModel = crossval.fit(training)

# 準(zhǔn)備測試未標(biāo)注的文件
test = spark.createDataFrame([
    (4, "spark i j k"),
    (5, "l m n"),
    (6, "mapreduce spark"),
    (7, "apache hadoop")
], ["id", "text"])

# 對測試文檔進(jìn)行預(yù)測, cvModel使用發(fā)現(xiàn)的最佳模型(lrModel)。
prediction = cvModel.transform(test)
selected = prediction.select("id", "text", "probability", "prediction")
for row in selected.collect():
    print(row)

訓(xùn)練驗證拆分

除了 CrossValidator 之外，Spark 還提供了用于超參數(shù)調(diào)優(yōu)的 TrainValidationSplit。TrainValidationSplit 只計算每個參數(shù)組合一次，而在 CrossValidator 的情況下是k次。因此，它的成本較低，但當(dāng)訓(xùn)練數(shù)據(jù)集不夠大時，它不會產(chǎn)生可靠的結(jié)果。

與 CrossValidator 不同，TrainValidationSplit 創(chuàng)建單個(訓(xùn)練、測試)數(shù)據(jù)集對。它使用 trainRatio 參數(shù)將數(shù)據(jù)集分成這兩部分。例如，當(dāng)trainRatio=0.75 時，TrainValidationSplit 將生成一個訓(xùn)練和測試數(shù)據(jù)集對，其中 75% 的數(shù)據(jù)用于訓(xùn)練，25% 用于驗證。

像 CrossValidator 一樣，TrainValidationSplit 最終使用最佳 ParamMap 和整個數(shù)據(jù)集匹配 Estimator。

from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.regression import LinearRegression
from pyspark.ml.tuning import ParamGridBuilder, TrainValidationSplit

# Prepare training and test data.
data = spark.read.format("libsvm")\
    .load("data/mllib/sample_linear_regression_data.txt")
train, test = data.randomSplit([0.9, 0.1], seed=12345)

lr = LinearRegression(maxIter=10)

# 我們使用ParamGridBuilder來構(gòu)造一個用于搜索的參數(shù)網(wǎng)格。
# TrainValidationSplit將嘗試所有值的組合，并使用評估器確定最佳模型。
paramGrid = ParamGridBuilder()\
    .addGrid(lr.regParam, [0.1, 0.01]) \
    .addGrid(lr.fitIntercept, [False, True])\
    .addGrid(lr.elasticNetParam, [0.0, 0.5, 1.0])\
    .build()

# 在這種情況下，估計器是簡單的線性回歸。
# TrainValidationSplit需要一個Estimator、一組Estimator ParamMaps 和一個 Evaluator。
tvs = TrainValidationSplit(estimator=lr,
                           estimatorParamMaps=paramGrid,
                           evaluator=RegressionEvaluator(),
                           # 80%的數(shù)據(jù)將用于培訓(xùn)，20%用于驗證。
                           trainRatio=0.8)

# 運(yùn)行TrainValidationSplit，并選擇最佳參數(shù)集。
model = tvs.fit(train)

# 對測試數(shù)據(jù)進(jìn)行預(yù)測。模型是參數(shù)組合后性能最好的模型。
model.transform(test)\
    .select("features", "label", "prediction")\
    .show()