【Apache Spark 】第 11 章使用 Apache Spark 管理、部署和扩展机器学习管道

图 11-1。MLflow 组件

Tracking（追踪）

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 模型的各种部署选项.

图 11-6。MLflow 模型注册表

近实时

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

笔记

在撰写本文时，分布式 XGBoost API 仅在 Java/Scala 中可用。本书的GitHub 存储库中包含了一个完整的示例。

笔记

如果工作人员在第一次加载模型权重后对其进行缓存，则后续调用相同 UDF 并加载相同模型的速度将显着加快。

Joblib

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

要使用 Joblib，请通过pip install joblibspark. 确保您使用的是scikit-learn0.21 或更高版本以及pyspark2.4.4 或更高版本。这里展示了如何进行分布式交叉验证的示例，同样的方法也适用于分布式超参数调整：

# 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

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

• 将单机 Hyperopt 与分布式训练算法（例如 MLlib）一起使用

• SparkTrials将分布式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))

fmin()生成新的超参数配置以供您使用training_function并将它们传递给SparkTrials. SparkTrials在每个 Spark 执行器上并行运行这些训练任务的批次作为单任务 Spark 作业。当 Spark 任务完成后，它会将结果和相应的损失返回给驱动程序。Hyperopt 使用这些新结果为未来的任务计算更好的超参数配置。这允许大规模扩展超参数调整。MLflow 还与 Hyperopt集成，因此您可以跟踪作为超参数调整的一部分训练的所有模型的结果。

的一个重要参数SparkTrials是parallelism。这决定了同时评估的最大试验次数。如果parallelism=1，那么您正在按顺序训练每个模型，但是通过充分利用自适应算法，您可能会得到更好的模型。如果您设置parallelism=max_evals（要训练的模型总数），那么您只是在进行随机搜索。1和之间的任何数字max_evals都允许您在可扩展性和适应性之间进行权衡。默认情况下，parallelism设置为 Spark 执行器的数量。您还可以指定 atimeout来限制允许的最大秒数fmin()。

即使 MLlib 不适合您的问题，希望您能看到在任何机器学习任务中使用 Spark 的价值。

# 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