简介

在本文中，我将介绍部分 MLOps 工作流程（从项目初始化到模型实验）。

在这里，我不会详细介绍 ML，因为这会让它变得冗长。我的主要重点是数据版本化，以及使用 DVC 和 DVClive 进行模型实验。

什么是 DVC，我们为什么需要它？

DVC 是数据版本控制的缩写。它是一种旨在管理机器学习项目复杂性的工具。它侧重于对数据集和模型等大型文件进行版本控制。DVC 可与 Git 无缝协作，管理机器学习项目中的代码和数据。虽然 Git 在版本控制和跟踪源代码变更方面表现出色，但在高效处理大型文件方面可能会遇到困难。DVC 与 Git 集成，克服了这一瓶颈。

Git 与 DVC 协作

Git 可追踪变更并管理机器学习项目代码库的演变。DVC 在 Git 仓库中存储轻量级元文件。这些文件包含与项目相关的大型数据集、模型和其他二进制文件的信息。实际数据文件存储在 Git 仓库之外，由 DVC 处理这些大文件的版本和链接。DVC 使用 Git 管理代码和轻量级元数据，而实际数据文件则高效地存储在单独的远程位置，如 Google Drive、Amazon S3 Bucket 和 Azure Blob Storage。

使用 cookiecutter 创建项目模板 

饼干切割器是一种模板，可以帮助我们快速创建相同形状的饼干。在编程领域，Cookiecutter 模板与之类似。它就像一个可重复使用的模具，用于建立新项目。与其每次都从头开始一个项目，我们可以使用 Cookiecutter 模板来快速创建一个包含所需文件和文件夹的基本结构。这不仅节省了时间，还确保了项目的一致性，就像饼干切割器可以帮助我们制作统一的饼干一样。

激活虚拟环境，进入终端，执行下面的命令 ：

cookiecutter -c v1 https://github.com/drivendata/cookiecutter-data-science

它会询问项目的基本细节，仅此而已。

该模板可创建出色的文件夹结构。请看快照：

更新 cookiecutter 模板

根据我们的项目要求，我们需要添加一些文件和目录。例如：

• params.yaml - 用于存储我们在项目中将进一步使用的所有参数。
• dvc.yaml - 在这里我们将以阶段的形式编写 e2e 执行工作流。
  
• temp dir - DVC 将使用它作为本地远程存储，以存储我们的数据集和 ML 模型。我们可以给目录起任何名字。

创建临时目录后，必须使用 DVC 将其设置为远程存储。这样，DVC 就能利用它来存储数据集和模型。除了本地存储外，还可以使用 Google Drive、Amazon S3 Bucket 和 Azure Blob Storage 等远程存储选项。只需运行以下命令即可配置本地远程存储：

cmd： "dvc remote add -d any_name temp/"

此处

any_name：你想赋予此远程存储的名称。用一个有意义的名称替换它。

temp/： 远程存储的 URL 或路径。在本例中，它是位于根目录下名为 temp/ 的本地目录。

我们不想跟踪 temp 文件夹（DVC 的本地远程存储）的更改，也不想在版本控制系统中包含该文件夹中的任何内容，因此我们要将 temp/ 添加到 .gitignore 中。

# create yaml file for parameters
touch params.yaml
# create yaml file for writing stages
touch dvc.yaml
# create temp folder, can use any name instead of temp
mkdir temp
# my_remote is remote name used to identify the remote. Must be unique.
dvc remote add -d my_remote directory/path/ 

Git 和 DVC 初始化

完成前面的步骤后，我们来初始化 Git 和 DVC。

git init     # initialize gitinit     # initialize git
dvc init     # intialize dvc

现在在 GitHub 上创建一个版本库，然后运行下面的命令将本地版本库与远程版本库链接起来。

git remote add origin https://github.com/user_name/remote_repo_name.git
git branch -M main
git push -u origin main

了解 "params.yaml "和 "dvc.yaml" 

params.yaml 是一个简单的配置文件，其中包含代码用来控制机器学习项目各个方面的参数和设置。我们可以放入超参数、文件路径、数据源、配置标志和实验标识符。这些参数可以在不更改代码的情况下轻松修改，为实验提供了灵活性和可重复性。

# file: params.yaml
base:
  project: creditcard-project
  target_col: Class
data_source:
  drive: https://drive.google.com/google_drive_link_where_data_is_stored
load_dataset:
  raw_data: /data/raw
  file_name: creditcard
make_dataset:
  test_split: 0.3
  random_state: 42
  processed_data: /data/processed
train_model:
  seed: 42
  n_estimators: 15
  max_depth: 8
  model_loc: /models

在 dvc.yaml 中，我们将定义完整的机器学习工作流程。从数据获取到模型评估。首先，我们将把工作流程划分为多个阶段，然后开始逐一处理，并将它们相互拼接，从而形成一个完整的管道。

# file: dvc.yaml
stages:
  load_dataset:
    cmd: python ./src/data/load_dataset.py
    deps:
    - ./src/data/load_dataset.py
    params:
    - data_source.drive
    - load_dataset.raw_data
    - load_dataset.file_name
    outs:
    # way to read parameters from params.yaml 
    - .${load_dataset.raw_data}/${load_dataset.file_name}.csv     # dump raw data at some loc
  make_dataset:
    cmd: python ./src/data/make_dataset.py
    deps:
    - ./src/data/make_dataset.py
    - .${load_dataset.raw_data}/${load_dataset.file_name}.csv     # depend on raw data dumped in previous stage
    params:
    - make_dataset.test_split
    - make_dataset.random_state
    - make_dataset.processed_data
    - load_dataset.raw_data
    - load_dataset.file_name
    outs:
    - .${make_dataset.processed_data}/train.csv
    - .${make_dataset.processed_data}/test.csv
  train_model:
    cmd: python ./src/models/train_model.py
    deps:
    - .${make_dataset.processed_data}/train.csv     # depend on train.csv 
    - ./src/models/train_model.py
    params:
    - train_model.seed
    - train_model.n_estimators
    - train_model.max_depth
    - train_model.model_loc
    - make_dataset.processed_data
    - base.target_col
    outs:
    - .${train_model.model_loc}/model.joblib
  predict_model:
    cmd: python ./src/models/predict_model.py
    deps:
    - ./src/models/predict_model.py
    - .${make_dataset.processed_data}/test.csv
    - .${train_model.model_loc}/model.joblib     # depend on model.joblib generated by previous stage
    params:
    - train_model.model_loc
    - make_dataset.processed_data
    - base.target_col
params:
- dvclive/params.yaml
metrics:
- dvclive/metrics.json
plots:
- dvclive/plots/metrics:
    x: step
# see every stage is consuming/depends on output of previous stage, this is how we stitch the stages and form a pipeline. 

让我们创建阶段

实施日志记录器

日志记录有助于提高代码的可维护性、加快问题的解决速度和项目的整体可靠性。它有助于调试、错误修复、故障排除、理解流程、性能监控等等。由于我们在这个项目中要处理多个文件，因此很难追查到任何错误或 bug，因此最好的做法是记录工作流程。实施一个日志记录器来记录工作流程并将其保存在某个位置。

# file: logger.py
import logging
import pathlib
from datetime import datetime
infologger = logging.getLogger(__name__)
infologger.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s : %(levelname)s : %(name)s : %(message)s')
log_file = f'{datetime.now().strftime("%d%b%y-%H.%M.%S")}.log'
# get current path and goto root 
log_dir_path = (pathlib.Path(__file__).parent.parent.as_posix() + '/logs')    
# create the directory if not available
pathlib.Path(log_dir_path).mkdir(parents = True, exist_ok = True)
log_file_path = pathlib.Path(log_dir_path + f'/{log_file}')
file_handler = logging.FileHandler(log_file_path)
file_handler.setFormatter(formatter)
infologger.addHandler(file_handler)
if __name__ == "__main__" :
     logging.info('Testing log')

下面我添加了一些日志文件的快照。

第一阶段--从 Google Drive 加载数据

让我们用代码从 Google Drive 加载数据集，并将其保存在 data/raw 目录中。我使用的是信用卡数据集，你可以从这里下载并保存在 Google Drive 中，然后将其公开。

# file: load_dataset.py
import pathlib
import yaml
import pandas as pd
from sklearn.model_selection import train_test_split
from src.log_config import infologger   
infologger.info('Executing : load_dataset.py')
infologger.info('purpose: fetch data from google drive and store it in data/raw dir')
def load_data(remote_loc) :
    # Load your dataset from a given path
     try : 
          infologger.info(f'data source {remote_loc}')
          remote_loc = 'https://drive.google.com/uc?id=' + remote_loc.split('/')[-2]
          df = pd.read_csv(remote_loc)
          return df
     except Exception as e : 
          infologger.info(f'Loading data from remote location failed with error : {e}')

def save_data(raw_data, output_path, file_name) : 
     # store data in data/raw dir
     try : 
          raw_data.to_csv(output_path + f'/{file_name}.csv', index = False)
          infologger.info(f'raw data saved suuccessfully at {output_path}')
     except Exception as e :
          infologger.info(f'Not able to save data. Error {e} caught.')
def main() : 
     curr_dir = pathlib.Path(__file__)    # at load_dataset.py
     home_dir = curr_dir.parent.parent.parent    # at creditcard_dvc
     param_file = home_dir.as_posix() + '/params.yaml'    # read from home dir
     params = yaml.safe_load(open(param_file))
     # input_file = sys.argv[1]
     remote_loc = params['data_source']['drive']
     output_path = home_dir.as_posix() + params['load_dataset']['raw_data']
     pathlib.Path(output_path).mkdir(parents = True, exist_ok = True)
     file_name = params['load_dataset']['file_name'] 
     data = load_data(remote_loc = remote_loc)
     save_data(data, output_path, file_name)
if __name__ == "__main__" : 
     main()

# params.yaml
base:
  project: creditcard-project
  target_col: Class
data_source:
  drive: https://drive.google.com/...
load_dataset:
  raw_data: /data/raw
  file_name: creditcard

# dvc.yaml
stages:
  load_dataset:
    cmd: python ./src/data/load_dataset.py
    deps:
    - ./src/data/load_dataset.py
    params:
    - data_source.drive
    - load_dataset.raw_data
    - load_dataset.file_name
    outs:
    # way to read parameters from params.yaml
    - .${load_dataset.raw_data}/${load_dataset.file_name}.csv    

继续在 dvc.yaml 中逐步添加参数和阶段。这将帮助你更好地理解工作流程和拼接阶段。请按照以下步骤操作：

- 首先在 Git 上推送更新后的代码库。

- 运行 dvc repro 执行阶段。

- 执行成功后，在 Git 上推送更新后的 dvc.lock 和 .gitignore 文件。请提供相关的提交信息。

- 如果数据集或模型中发生任何更改，请使用 dvc push 通过 DVC 跟踪更改。

下面是输出结果的快照。

此外，还要提供合适的提交语句，以免将来遇到麻烦。

第二阶段--执行训练-测试分割

让我们执行代码，从 data/raw 目录中获取数据，对其执行训练-测试分割，并将其存储到 data/processed 目录中。

# file: make_dataset.py
import pathlib
import yaml
import pandas as pd
from sklearn.model_selection import train_test_split
from src.log_config import infologger   
infologger.info('Executing : make_dataset.py')
infologger.info('purpose: fetch data from data/raw and perform train-test split')
def load_data(data_path) :
    # Load your dataset from a given path
    try : 
        infologger.info(f'data source {data_path}')
        data = pd.read_csv(data_path)
        infologger.info('data loaded successfully')
        return data
    except Exception as e : 
        infologger.info(f'encountered error {e} while loading data')
def split_data(df, test_split, seed) :
    # Split the dataset into train and test sets
    try : 
        train, test = train_test_split(df, test_size = test_split, random_state = seed)
        infologger.info(f'perform train-test split with test_split = {test_split} and seed = {seed}')
        return train, test
    except Exception as e : 
        infologger.info(f'encountered error {e} while spliting data')
def save_data(train, test, output_path) :
    # Save the split datasets to the specified output path
    try : 
        train.to_csv(output_path + '/train.csv', index = False)
        test.to_csv(output_path + '/test.csv', index = False)
        infologger.info(f'splited data saved suuccessfully at {output_path}')
    except Exception as e : 
        infologger.info(f'encountered error {e} while saving the splited data')
def main() :
    curr_dir = pathlib.Path(__file__)
    home_dir = curr_dir.parent.parent.parent
    param_file = home_dir.as_posix() + '/params.yaml'
    params = yaml.safe_load(open(param_file))
    # input_file = sys.argv[1]
    data_path = home_dir.as_posix() + params['load_dataset']['raw_data'] 
    output_path = home_dir.as_posix() + params['make_dataset']['processed_data']
    pathlib.Path(output_path).mkdir(parents = True, exist_ok = True)
    data_file = f"{data_path}/{params['load_dataset']['file_name']}.csv"
    data = load_data(data_path = data_file)
    train_data, test_data = split_data(data, params['make_dataset']['test_split'], params['make_dataset']['random_state'])
    save_data(train_data, test_data, output_path)
if __name__ == "__main__" :
    main()  

# params.yaml
# add below section in params.yaml
make_dataset:
  test_split: 0.3
  random_state: 42
  processed_data: /data/processed

# dvc.yaml
# add below stage just after load_dataset stage under "stages:" \
# section of dvc.yaml
make_dataset:
  cmd: python ./src/data/make_dataset.py
  deps:
  - ./src/data/make_dataset.py
  - .${load_dataset.raw_data}/${load_dataset.file_name}.csv     # depend on raw data dumped in previous stage
  params:
  - make_dataset.test_split
  - make_dataset.random_state
  - make_dataset.processed_data
  - load_dataset.raw_data
  - load_dataset.file_name
  outs:
  - .${make_dataset.processed_data}/train.csv
  - .${make_dataset.processed_data}/test.csv

再次重复同样的步骤：

- 首先在 Git 上推送更新后的代码库

- 运行 dvc repro

- 执行成功后，在 Git 上推送更新后的 dvc.lock 、.gitignore 和 dvclive（如果有）文件夹。添加有意义的提交信息。

- 使用 dvc push 通过 DVC 跟踪更改 就是这样。

下面是输出结果的照片。

第三阶段--训练机器学习模型

我们提取了数据，进行了拆分，现在就可以训练模型了。让我们来训练模型，看看 DVC 的神奇之处。

# file: train_model.py
import pathlib 
import yaml
import joblib
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn import metrics
from src.log_config import infologger
from dvclive import Live
infologger.info('Executing : train_model.py')
infologger.info('purpose: train the model')
def train_model(train_features, target, n_estimators, max_depth, seed) :
     try : 
          model = RandomForestClassifier(n_estimators = n_estimators, max_depth = max_depth, random_state = seed)
          infologger.info(f'training {type(model).__name__} model')
          model.fit(train_features, target)
          pred_train = model.predict(train_features)
          accuracy = metrics.accuracy_score(target, pred_train)    # training accuracy
          try : 
               with Live(resume = True) as live :
                    # log model parameters
                    live.log_param('n_estimator', n_estimators)
                    live.log_param('max_depth', max_depth)
                    live.log_param('random_state', seed)
                    # log training metrics
                    live.log_metric('training/accuracy', float("{:.2f}".format(accuracy)))
                    live.log_metric('training/precision', float("{:.2f}".format(metrics.\
                                                                                precision_score(target, pred_train, zero_division = 1))))
                    live.log_metric('training/recall', float("{:.2f}".format(metrics.recall_score(target, pred_train))))
               infologger.info('successfully logged parameters & metrics through dvclive')
          except Exception as ie : 
               infologger.info(f'failed to load dvclive, encountered error {ie}')
          return model
     except Exception as oe : 
          infologger.info(f'failed to load model, encountered error {oe}')
          
          
def save_model(model, output_path) : 
     # Save the trained model to the specified output path
     try : 
          joblib.dump(model, output_path + '/model.joblib')
          infologger.info(f'model saved successfully at loc {output_path}')
     except Exception as e : 
          infologger.info(f'failed to dump the model, encountered error {e}')
def main() :
     curr_dir = pathlib.Path(__file__)
     home_dir = curr_dir.parent.parent.parent
     params_file = home_dir.as_posix() + '/params.yaml' 
     params = yaml.safe_load(open(params_file))
  
     # input_file = sys.argv[1]
     data_path = home_dir.as_posix() + params['make_dataset']['processed_data']
     output_path = home_dir.as_posix() + params['train_model']['model_loc']
     pathlib.Path(output_path).mkdir(parents = True, exist_ok = True)
     TARGET = params['base']['target_col']
     train_features = pd.read_csv(data_path + '/train.csv')
     X = train_features.drop(TARGET, axis = 1)
     y = train_features[TARGET]
     trained_model = train_model(X, y, params['train_model']['n_estimators'], params['train_model']['max_depth'],
                                 params['train_model']['seed'])
     save_model(trained_model, output_path)

if __name__ == "__main__" :
     main()

# params.yaml
# add below code in params.yaml
train_model:
  seed: 42
  n_estimators: 15
  max_depth: 8
  model_loc: /models

# dvc.yaml
# add below stage just after make_dataset stage under "stages:" \
# section of dvc.yaml
train_model:
  cmd: python ./src/models/train_model.py
  deps:
  - .${make_dataset.processed_data}/train.csv     # depend on train.csv &
  - ./src/models/train_model.py
  params:
  - train_model.seed
  - train_model.n_estimators
  - train_model.max_depth
  - train_model.model_loc
  - make_dataset.processed_data
  - base.target_col
  outs:
  - .${train_model.model_loc}/model.joblib

在 train_model.py 中，你会发现我使用了 dvclive。让我来解释一下它的神奇之处：

- dvclive 允许我们跟踪和记录机器学习实验的指标。这包括准确率、损失、精确度、召回率、混淆度量等指标，或其他任何你想监控的自定义指标。

- 它与 DVC 无缝集成，确保实验指标与代码和数据一起存储和版本化。

- 它能在训练或评估过程中提供指标的实时可视化。这有助于监控模型的进展，并及时发现趋势或问题。

- 它提供了一个基于网络的界面或仪表板，我们可以在其中查看和分析记录的指标。这可以提高实验的可解释性。

- 它还有助于在不同的运行中保持一致的实验日志记录方法，从而更容易比较结果并了解代码或数据变化的影响。

当我们执行这段代码时，它会将日志数据存储在传递给 Live() 的目录 (dir) 下。如果没有提供，则默认使用 dvclive。

dvclive 包中有多种方法可用于跟踪参数、度量或绘图，这里我使用了其中的两个 log_param() 和 log_metric() 添加了 dvclive 的照片。

- log_param() 会在 dvclive 文件夹下创建 params.yaml 文件，并跟踪 log_param() 记录的所有参数。我们可以运行 dvc params diff 命令，比较当前工作区（未提交）和上一个实验（已提交）中发生的参数更改。我们还可以使用任何实验的提交 ID 比较它们之间的变化，只需使用 dvc params commit_id1 commit_id2 命令即可。它只会比较之前提交的实验和最近未提交的实验之间的变化。

- log_metric() 会在 dvclive 文件夹下创建 metrics.json 文件，并跟踪 log_metric() 记录的所有指标。在这里，我们可以使用一些命令来比较指标：

# it will show metrics of curr experiment
dvc metrics show
# compare metrics of curr. exp and previous workspace
# only show metrics which get changed
dvc metrics diff
# compare metrics of different commits
dvc metrics commit_id1 commit_id2
# show all metrics irresp. of their changes
dvc metrics --all

DVClive 将参数/指标保存在 yaml 和 json 文件以及 .tsv 文件中。params.yaml 和 metrics.json 将保存最新日志，而 .tsv 文件将保存所有日志。它会在我们进行实验时不断添加数据。但 yaml 和 json 文件会在每次实验后覆盖数据。

完成代码后，我们就可以按照上述步骤跟踪代码库和数据的变化。

下面我添加了输出结果的照片。

第四阶段--评估模型

让我们来评估一下我们的模型，看看它处于什么位置。

# file: predict_model.py
import yaml
import joblib
import pathlib
import pandas as pd
from dvclive import Live
from sklearn import metrics
from src.log_config import infologger
infologger.info('Executing : predict_model.py')
infologger.info('purpose: evaluate the model and log the data')

def load_model(model_path) : 
     # load model and return it
     try : 
          with open(model_path + '/model.joblib', 'rb') as f :
               model = joblib.load(f)
          return model
     except Exception as e : 
          infologger.info(f'failed to load model, encountered error {e}')

def evaluate(data, model_path) :
     try : 
          model = load_model(model_path = model_path)
          infologger.info(f'model loaded successfully from path: {model_path}')
     except Exception as e : 
          infologger.info(f'failed to load model, encountered error {e}')
     else : 
          TARGET = [params['base']['target_col']]
          X = data.drop(TARGET, axis = 1)
          y = data[TARGET]
          predictions_by_class = model.predict_proba(X)
          y_pred = model.predict(X)
          predictions = predictions_by_class[:, 1]
          try : 
          # track all using live
               with Live(resume = True) as live :
                    live.log_metric('testing/roc_auc_score', float("{:.2f}".format(metrics.roc_auc_score(y, predictions))))
                    live.log_metric('testing/bal_acc_score', float("{:.2f}".format(metrics.balanced_accuracy_score(y, y_pred))))
                    live.log_metric('testing/recall', float("{:.2f}".format(metrics.recall_score(y, y_pred))))
                    live.log_metric('testing/precision', float("{:.2f}".format(metrics.precision_score(y, y_pred, zero_division = 1))))
               
               infologger.info('testing metrics logged successfully')
          except Exception as e : 
               infologger.info(f'failed to log metrics, encountered error {e}')

if __name__ == '__main__' : 
     curr_dir = pathlib.Path(__file__)
     home_dir = curr_dir.parent.parent.parent
     params_file = home_dir.as_posix() + '/params.yaml' 
     params = yaml.safe_load(open(params_file))
     model_path = home_dir.as_posix() + params['train_model']['model_loc']
     test_data = pd.read_csv(f"{home_dir.as_posix()}{params['make_dataset']['processed_data']}/test.csv")
     # Evaluate test datasets.
     evaluate(test_data, model_path)

# dvc.yaml
# add below stage just after train_model stage under "stages:" \
# section of dvc.yaml  
predict_model:
    cmd: python ./src/models/predict_model.py
    deps:
    - ./src/models/predict_model.py
    # my thought is, if train.csv changes then we'll run train_model bt if test.csv changes we'll do prediction
    - .${make_dataset.processed_data}/test.csv
    - .${train_model.model_loc}/model.joblib     # depend on model.joblib generated by previous stage
    params:
    - train_model.model_loc
    - make_dataset.processed_data
    - base.target_col

我们没有使用任何外部参数，因此没有更新 params.yaml。要在 Git 和 DVC 上推送更改，请按照上述步骤操作。

下面我添加了输出结果的照片。

dvc.lock" 文件

DVC 还创建了自己的暂存文件，并将其命名为 dvc.lock。这个文件看起来与我们的 dvc.yaml 文件非常相似，唯一不同的是，它还在每个文件名后面存储了哈希码和文件大小，以便 Git 和 DVC 能同步。我添加了一张照片：

dvc.lock 文件存储了哈希码，而相同的哈希码被标记到文件中。每次发生任何更改时，DVC 都会在远程存储器（这里是临时文件夹）中存储一份副本。在每次更改/实验后，我们都会跟踪 dvc.lock 文件，现在假设你检出了任何提交，它就会回到 dvc.lock 文件的快照，当我们进行 dvc pull 时，它就会拉出 dvc.lock 快照中提到的相同哈希码文件。事情就是这样的。

缝合阶段 

我在这里添加了 dvc.yaml 的实际代码。在 stages: 部分，你会发现每个阶段都依赖于前一个阶段的输出 (outs:)。各阶段通过 deps 和 outs 连接，这就是我们拼接各阶段并构建完整工作流程的方式。在执行实验时，如果 deps: 发生任何变化，则执行该阶段，否则跳过该阶段。这就是 DVC 的魅力所在。

运行实验️ 

一旦我们完成了代码库，现在就可以轻松更改 params.yaml 中的任何参数并执行管道。你可以通过终端比较指标，或者点击 DVC 并选择 "显示实验"（Show Experiments），这将打开 "实验"（Experiments）选项卡，在这里你将获得从文件的参数到哈希代码（由 DVC 分配）的所有信息。下面我添加了实验选项卡的照片。