4. 配置MLflow Tracking Server

任务目标

在MLOps工作流中，数据集与代码一样需要版本管理。本案例使用DVC（Data Version Control）对本地数据集目录进行版本化跟踪，实现数据快照、版本切换与远程存储同步，解决数据集迭代后无法回溯、协作时传输原始文件的问题。

环境与依赖

  • Python 3.8+，已安装 dvc（pip install dvc）与 git
  • 本案例使用本地目录模拟远程存储（/tmp/dvc-remote），无需云服务账号

实现步骤与代码

以下脚本完整演示：初始化DVC、添加数据目录、修改数据后提交新版本、切换历史版本、推送至远程存储的全过程。所有命令均通过Python的subprocess执行，便于在Notebook或CI中直接运行验证。

import os
import subprocess
import shutil

# ---------- 0. 准备实验目录 ----------
WORK = "/tmp/dvc_demo"
DATA = os.path.join(WORK, "data")
REMOTE = "/tmp/dvc-remote"

for p in [WORK, REMOTE]:
    shutil.rmtree(p, ignore_errors=True)
    os.makedirs(p)

os.chdir(WORK)
subprocess.run(["git", "init", "-q"])
subprocess.run(["dvc", "init", "-q"])

# ---------- 1. 创建初始数据版本 v1 ----------
os.makedirs(DATA)
with open(os.path.join(DATA, "train.csv"), "w") as f:
    f.write("id,feature,label\n1,0.5,cat\n2,1.2,dog\n")
with open(os.path.join(DATA, "meta.json"), "w") as f:
    f.write('{"version": "v1", "rows": 2}\n')

subprocess.run(["dvc", "add", "data"], check=True, capture_output=True)
subprocess.run(["git", "add", "-A"], check=True)
subprocess.run(["git", "commit", "-m", "data v1: 2 rows"], check=True, capture_output=True)

# ---------- 2. 修改数据生成 v2 ----------
with open(os.path.join(DATA, "train.csv"), "a") as f:
    f.write("3,0.8,bird\n4,1.5,cat\n")

subprocess.run(["dvc", "add", "data"], check=True, capture_output=True)
subprocess.run(["git", "add", "-A"], check=True)
subprocess.run(["git", "commit", "-m", "data v2: 4 rows"], check=True, capture_output=True)

# ---------- 3. 配置并推送远程存储 ----------
subprocess.run(["dvc", "remote", "add", "-d", "myremote", REMOTE], check=True, capture_output=True)
subprocess.run(["dvc", "push"], check=True, capture_output=True)

# ---------- 4. 验证：切换回 v1 并确认内容 ----------
subprocess.run(["git", "checkout", "-q", "HEAD~1"])
subprocess.run(["dvc", "checkout"], check=True, capture_output=True)

with open(os.path.join(DATA, "train.csv")) as f:
    v1_content = f.read()
print("=== v1 train.csv ===")
print(v1_content)

# ---------- 5. 切回最新版本并确认 ----------
subprocess.run(["git", "checkout", "-q", "master"])
subprocess.run(["dvc", "checkout"], check=True, capture_output=True)

with open(os.path.join(DATA, "train.csv")) as f:
    v2_content = f.read()
print("=== v2 train.csv ===")
print(v2_content)

# ---------- 6. 验证远程存储中已有缓存文件 ----------
remote_files = os.listdir(os.path.join(REMOTE, "files", "md5"))
print(f"远程缓存文件数: {len(remote_files)}")

运行方式

将上述代码保存为 dvc_demo.py，在终端执行：

python dvc_demo.py

验证方法与判定标准

  • 版本切换正确性：程序输出中，v1 train.csv应只包含2行数据（id为1、2），v2 train.csv应包含4行数据（id为1～4）。若内容与上述一致，说明DVC正确跟踪并恢复了不同版本的数据快照。
  • 远程同步成功：最后一行输出“远程缓存文件数”大于0，表示数据已推送至/tmp/dvc-remote。可进一步在终端执行 dvc status，若输出“Data and pipelines are up to date.”，则本地与远程完全同步。
  • Git与DVC协作：执行 git log --oneline 应看到两条提交记录（v1与v2），执行 dvc log 可查看数据文件的版本历史。

常见问题

  • dvc init 报错“Git is not installed”：先执行 git init 并完成一次提交后再运行 dvc init。
  • dvc checkout 后文件未变化：确认当前Git HEAD对应的 .dvc 文件已提交；若未提交，DVC无法知道该版本的数据指针。
  • 远程存储路径不存在：DVC不会自动创建远程目录，需先 os.makedirs(REMOTE, exist_ok=True)（本案例已包含）。
← 上一章 📖 返回目录 下一章 →