MLflow Tracking 是一个日志 API，它与实际进行训练的库和环境无关。它是围绕运行的概念组织的，运行是数据科学代码的执行。运行被聚合到实验中，因此许多运行可以成为给定实验的一部分。

MLflow 跟踪服务器可以托管许多实验。您可以使用笔记本、本地应用程序或云作业登录到跟踪服务器，如图 11-2所示。

让我们检查一些可以记录到跟踪服务器的内容:

Parameters

代码的键/值输入——例如，像随机森林num_trees或max_depth在随机森林中的超参数

Metrics

数值（可以随时间更新）——例如，RMSE 或精度值

Artifacts

文件、数据和模型——例如matplotlib图像或 Parquet 文件

Metadata

有关运行的信息，例如执行运行的源代码或代码的版本（例如，代码版本的 Git 提交哈希字符串）

Models

您训练的模型

默认情况下，跟踪服务器将所有内容记录到文件系统中，但您可以指定一个数据库以加快查询速度，例如参数和指标。让我们将 MLflow 跟踪添加到第 10 章中的随机森林代码中：

# In Python
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, VectorAssembler
from pyspark.ml.regression import RandomForestRegressor
from pyspark.ml.evaluation import RegressionEvaluator

filePath = """/databricks-datasets/learning-spark-v2/sf-airbnb/
sf-airbnb-clean.parquet"""
airbnbDF = spark.read.parquet(filePath)
(trainDF, testDF) = airbnbDF.randomSplit([.8, .2], seed=42)

categoricalCols = [field for (field, dataType) in trainDF.dtypes 
                   if dataType == "string"]
indexOutputCols = [x + "Index" for x in categoricalCols]
stringIndexer = StringIndexer(inputCols=categoricalCols, 
                              outputCols=indexOutputCols, 
                              handleInvalid="skip")

numericCols = [field for (field, dataType) in trainDF.dtypes 
               if ((dataType == "double") & (field != "price"))]
assemblerInputs = indexOutputCols + numericCols
vecAssembler = VectorAssembler(inputCols=assemblerInputs, 
                               outputCol="features")

rf = RandomForestRegressor(labelCol="price", maxBins=40, maxDepth=5, 
                           numTrees=100, seed=42)

pipeline = Pipeline(stages=[stringIndexer, vecAssembler, rf])

要开始使用 MLflow 进行日志记录，您需要使用mlflow.start_run(). mlflow.end_run()本章中的示例将使用with子句在块的末尾自动结束运行，而不是显式调用with：

# In Python 
import mlflow
import mlflow.spark
import pandas as pd

with mlflow.start_run(run_name="random-forest") as run:
  # Log params: num_trees and max_depth
  mlflow.log_param("num_trees", rf.getNumTrees())
  mlflow.log_param("max_depth", rf.getMaxDepth())
 
  # Log model
  pipelineModel = pipeline.fit(trainDF)
  mlflow.spark.log_model(pipelineModel, "model")

  # Log metrics: RMSE and R2
  predDF = pipelineModel.transform(testDF)
  regressionEvaluator = RegressionEvaluator(predictionCol="prediction", 
                                            labelCol="price")
  rmse = regressionEvaluator.setMetricName("rmse").evaluate(predDF)
  r2 = regressionEvaluator.setMetricName("r2").evaluate(predDF)
  mlflow.log_metrics({"rmse": rmse, "r2": r2})

  # Log artifact: feature importance scores
  rfModel = pipelineModel.stages[-1]
  pandasDF = (pd.DataFrame(list(zip(vecAssembler.getInputCols(), 
                                    rfModel.featureImportances)), 
                           columns=["feature", "importance"])
              .sort_values(by="importance", ascending=False))

  # First write to local filesystem, then tell MLflow where to find that file
  pandasDF.to_csv("feature-importance.csv", index=False)
  mlflow.log_artifact("feature-importance.csv")

让我们检查一下 MLflow UI，您可以通过mlflow ui在终端中运行并导航到http://localhost:5000/来访问它。图 11-3显示了 UI 的屏幕截图。

图 11-3。MLflow 用户界面

UI 存储给定实验的所有运行。您可以搜索所有运行，筛选符合特定条件的运行，并排比较运行等。如果您愿意，还可以将内容导出为 CSV 文件以在本地进行分析。在名为"random-forest". 您应该会看到如图 11-4 所示的屏幕。

您会注意到它会跟踪用于此 MLflow 运行的源代码，并存储所有相应的参数、指标等。您可以在自由文本中添加有关此运行的注释以及标签。运行完成后，您无法修改参数或指标。

您还可以使用MlflowClientREST API 或 REST API 查询跟踪服务器：

# In Python
from mlflow.tracking import MlflowClient

client = MlflowClient()
runs = client.search_runs(run.info.experiment_id, 
                          order_by=["attributes.start_time desc"], 
                          max_results=1)

run_id = runs[0].info.run_id
runs[0].data.metrics

这会产生以下输出：

{'r2'：0.22794251914574226，'rmse'：211.5096898777315}

我们将此代码作为MLflow 项目托管在本书的GitHubmax_depth存储库中，因此您可以尝试使用和的不同超参数值运行它num_trees。MLflow 项目中的 YAML 文件指定了库依赖项，因此此代码可以在其他环境中运行：

# In Python
mlflow.run(
  "https://github.com/databricks/LearningSparkV2/#mlflow-project-example", 
  parameters={"max_depth": 5, "num_trees": 100})

# Or on the command line
mlflow run https://github.com/databricks/LearningSparkV2/#mlflow-project-example
-P max_depth=5 -P num_trees=100

现在您已经跟踪并重现了您的实验，让我们讨论一下可用于您的 MLlib 模型的各种部署选项.

近实时

如果您的用例需要数百毫秒到几秒的预测，您可以构建一个使用 MLlib 生成预测的预测服务器。虽然这不是 Spark 的理想用例，因为您正在处理非常少量的数据，但与流式处理或批处理解决方案相比，您将获得更低的延迟。

根据其文档，Joblib是“一组在 Python 中提供轻量级流水线的工具”。它有一个 Spark 后端，用于在 Spark 集群上分发任务。Joblib 可用于超参数调整，因为它会自动将您的数据副本广播给所有工作人员，然后这些工作人员在其数据副本上创建具有不同超参数的自己的模型。这使您可以并行训练和评估多个模型。您仍然有一个基本限制，即单个模型和所有数据必须适合单个机器，但是您可以简单地并行化超参数搜索，如图 11-7所示。

# In Python
from sklearn.utils import parallel_backend
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
from sklearn.model_selection import GridSearchCV
import pandas as pd
from joblibspark import register_spark

register_spark() # Register Spark backend

df = pd.read_csv("/dbfs/databricks-datasets/learning-spark-v2/sf-airbnb/
  sf-airbnb-numeric.csv")
X_train, X_test, y_train, y_test = train_test_split(df.drop(["price"], axis=1), 
  df[["price"]].values.ravel(), random_state=42)

rf = RandomForestRegressor(random_state=42)
param_grid = {"max_depth": [2, 5, 10], "n_estimators": [20, 50, 100]}
gscv = GridSearchCV(rf, param_grid, cv=3)

with parallel_backend("spark", n_jobs=3):
  gscv.fit(X_train, y_train)
  
print(gscv.cv_results_)

有关从交叉验证器返回的参数的说明，请参阅scikit-learn GridSearchCV 文档。

Hyperopt是一个 Python 库，用于“对尴尬的搜索空间进行串行和并行优化，其中可能包括实值、离散和条件维度”。您可以通过pip install hyperopt. 使用 Apache Spark 扩展 Hyperopt有两种主要方法：

对于前一种情况，与任何其他库相比，您无需配置任何特殊配置即可将 MLlib 与 Hyperopt 结合使用。那么，让我们看看后一种情况：具有单节点模型的分布式 Hyperopt。不幸的是，在撰写本文时，您无法将分布式超参数评估与分布式训练模型结合起来。用于并行化Keras模型的超参数搜索的完整代码示例可以在本书的GitHub 存储库中找到；这里只包含一个片段来说明 Hyperopt 的关键组件：

# In Python
import hyperopt

best_hyperparameters = hyperopt.fmin(
  fn = training_function,
  space = search_space,
  algo = hyperopt.tpe.suggest,
  max_evals = 64,
  trials = hyperopt.SparkTrials(parallelism=4))

# In pandas
import pandas as pd
pdf = pd.read_csv(csv_path, header=0, sep=";", quotechar='"')
pdf["duration_new"] = pdf["duration"] + 100

# In koalas
import databricks.koalas as ks
kdf = ks.read_csv(file_path, header=0, sep=";", quotechar='"')
kdf["duration_new"] = kdf["duration"] + 100