机器学习全生命周期管理：MLflow实战指南

# 机器学习全生命周期管理：MLflow实战指南 ## 1. 机器学习面临的挑战 在机器学习项目中，数据科学家们常常面临着结果难以复现和模型缺乏标准化的问题。同一团队的数据科学家即使记录了代码和参数，也可能由于执行环境（如系统配置或库依赖）不同而无法复现彼此的结果。此外，不同团队在存储机器学习模型时可能使用不同的库和约定，这在团队间共享工作时会成为一个难题。 ## 2. MLflow简介 MLflow是一个用于管理端到端机器学习生命周期的开源平台，它有助于复现和共享实验、管理模型，并为最终用户部署模型。除了REST API和CLI外，它还提供了Python、R和Java/Scala的API。 MLflow有四个主要组件： |组件名称|功能描述| | ---- | ---- | |MLflow Tracking|记录参数、指标、代码版本、模型以及诸如绘图和文本等工件。| |MLflow Projects|提供一种可重用、可复现的格式，用于与其他数据科学家共享或转移到生产环境，帮助管理模型训练过程。| |MLflow Models|使你能够打包模型，以便部署到各种模型服务和推理平台，提供一致的API来加载和应用模型，而无需考虑构建模型所使用的底层库。| |MLflow Registry|使你能够在中央存储库中协同跟踪模型谱系、模型版本、阶段转换和注释。| ### 2.1 安装MLflow 使用pip安装MLflow非常简单： ```bash $ pip3 install mlflow ``` MLflow与许多流行的机器学习框架集成，如Spark、TensorFlow、PyTorch等。要导入特定于Spark的MLflow组件，只需运行`import mlflow.spark`。 ## 3. 实验跟踪 在典型的机器学习项目中，通常需要尝试多种算法和模型来解决问题，因此需要跟踪相关的数据集、超参数和指标。传统上，实验跟踪通常使用电子表格等临时工具，效率低下甚至不可靠。 MLflow Tracking是一个API和UI，用于在运行机器学习代码时记录参数、代码版本、指标和工件，并在后续可视化结果。你可以在任何环境（如独立脚本或笔记本）中使用MLflow Tracking将结果记录到本地文件或服务器，然后比较多次运行的结果。它与库无关，并与多个框架集成。 ### 3.1 MLflow Tracking的关键功能 - 基于实验的运行列表和比较 - 按参数或指标值搜索运行 - 可视化运行指标 - 下载运行结果 ### 3.2 将MLflow Tracking添加到决策树代码中 假设你已经下载了Covtype数据集并熟悉它。以下是在PySpark shell中添加MLflow Tracking到决策树代码的步骤： #### 3.2.1 准备数据和机器学习管道 ```python from pyspark.ml import Pipeline from pyspark.sql.functions import col from pyspark.sql.types import DoubleType from pyspark.ml.feature import VectorAssembler from pyspark.ml.classification import DecisionTreeClassifier data_without_header = spark.read.option("inferSchema", True).\ option("header", False).\ csv("data/covtype.data") colnames = ["Elevation", "Aspect", "Slope", "Horizontal_Distance_To_Hydrology", "Vertical_Distance_To_Hydrology", "Horizontal_Distance_To_Roadways", "Hillshade_9am", "Hillshade_Noon", "Hillshade_3pm", "Horizontal_Distance_To_Fire_Points"] + \ [f"Wilderness_Area_{i}" for i in range(4)] + \ [f"Soil_Type_{i}" for i in range(40)] + \ ["Cover_Type"] data = data_without_header.toDF(*colnames).\ withColumn("Cover_Type", col("Cover_Type").\ cast(DoubleType())) (train_data, test_data) = data.randomSplit([0.9, 0.1]) input_cols = colnames[:-1] vector_assembler = VectorAssembler(inputCols=input_cols,outputCol="featureVector") classifier = DecisionTreeClassifier(seed = 1234, labelCol="Cover_Type", featuresCol="featureVector", predictionCol="prediction") pipeline = Pipeline(stages=[vector_assembler, classifier]) ``` #### 3.2.2 开始使用MLflow记录 ```python import mlflow import mlflow.spark import pandas as pd from pyspark.ml.evaluation import MulticlassClassificationEvaluator with mlflow.start_run(run_name="decision-tree"): # Log param: max_depth mlflow.log_param("max_depth", classifier.getMaxDepth()) # Log model pipeline_model = pipeline.fit(train_data) mlflow.spark.log_model(pipeline_model, "model") # Log metrics: Accuracy and F1 pred_df = pipeline_model.transform(test_data) evaluator = MulticlassClassificationEvaluator(labelCol="Cover_Type", ```