7.流水线构建与执行

在机器学习项目中，代码只是整个故事的一部分。真正让项目运转起来的是数据流转的过程：原始数据经过清洗、特征工程、模型训练，最终产出结果。这个过程就像一条生产线，每个环节都依赖前一个环节的输出。DVC 的流水线功能就是为这条生产线设计的，它让我们能够用代码的方式定义、执行和复现整个工作流程。

Stage 添加与修改

构建流水线的第一步是定义 Stage。每个 Stage 代表一个独立的处理步骤，通常是一个脚本或命令，加上它的输入依赖和输出文件。DVC 提供了两种方式来管理 Stage：命令行工具和直接编辑配置文件。

使用 dvc stage add 创建 Stage

dvc stage add 是最直接的创建方式。它把命令、依赖和输出打包成一个 Stage，写入 dvc.yaml 文件。看个简单的例子：假设我们有一个文本文件，想统计其中的数字个数。

$ dvc stage add -n filter \
                -d text.txt \
                -o numbers.txt \
                "cat text.txt | egrep '[0-9]+' > numbers.txt"

这条命令创建了一个名为 filter 的 Stage。-n 指定名称，-d 指定依赖文件，-o 指定输出文件，最后是要执行的命令。执行后，dvc.yaml 文件中会多出这样的定义：

stages:
  filter:
    cmd: cat text.txt | egrep '[0-9]+' > numbers.txt
    deps:
      - text.txt
    outs:
      - numbers.txt

创建第二个 Stage 来处理过滤后的数字：

$ dvc stage add -n count \
                -d numbers.txt \
                -d process.py \
                -M count.txt \
                "python process.py numbers.txt > count.txt"

这里用 -M 标记 count.txt 为指标文件，这类文件通常很小，可以直接用 Git 追踪，不需要 DVC 缓存。现在 dvc.yaml 包含两个 Stage，DVC 会自动识别它们之间的依赖关系——count 依赖 filter 的输出 numbers.txt。

手动编辑 dvc.yaml

虽然 dvc stage add 很方便，但直接编辑 dvc.yaml 更灵活。YAML 格式清晰易读，支持版本控制，也便于代码审查。上面的例子手动写成：

stages:
  filter:
    cmd: cat text.txt | egrep '[0-9]+' > numbers.txt
    deps:
      - text.txt
    outs:
      - numbers.txt
  count:
    cmd: python process.py numbers.txt > count.txt
    deps:
      - numbers.txt
      - process.py
    outs:
      - count.txt

手动编辑的优势在于可以使用一些高级功能，比如模板化配置，这是命令行工具不支持的。团队协同时，直接在 YAML 文件中修改也更容易追踪变更。

修改现有 Stage

项目迭代过程中，Stage 的定义经常需要调整。比如修改训练脚本的学习率参数，或者增加新的数据依赖。这时可以用 -f 强制覆盖已有 Stage：

$ dvc stage add -n train \
                -f \
                -d train_model.py \
                -d matrix-train.p \
                -o model.p \
                python train_model.py 18494003 model.p

-f 会完全替换同名的 Stage 定义。如果只想添加依赖或输出，而不想重新执行 Stage，可以直接编辑 dvc.yaml，然后运行 dvc commit 更新状态。假设 prepare Stage 已经执行过，后来发现漏掉了 data/raw.csv 依赖和 data/validate 输出：

stages:
  prepare:
    cmd: python src/prepare.py
    deps:
      - src/prepare.py
      - data/raw.csv  # 新增的依赖
    outs:
      - data/train
      - data/validate  # 新增的输出

如果 data/raw.csv 和 data/validate 已经存在于工作区，运行 dvc commit 就能把这些新指定的输出加入缓存，同时更新 dvc.lock 中的哈希值，无需重新执行耗时的 prepare 阶段。

流水线执行 repro

定义好 Stage 后，下一步是让流水线运转起来。dvc repro 是核心命令，它分析依赖图，智能决定哪些 Stage 需要执行，然后按正确顺序运行它们。

基本执行流程

执行 dvc repro 时，DVC 会读取所有 dvc.yaml 文件，构建完整的依赖图（DAG）。然后检查每个 Stage 的状态：

1. 如果依赖文件的内容哈希值发生变化，该 Stage 需要重新执行
2. 如果输出文件缺失，该 Stage 需要重新执行
3. 如果依赖和输出都没变化，跳过该 Stage

看个完整的例子。假设我们已经定义了 filter 和 count 两个 Stage，第一次执行：

$ dvc repro
Running stage 'filter':
> cat text.txt | egrep '[0-9]+' > numbers.txt
Generating lock file 'dvc.lock'
Updating lock file 'dvc.lock'

Running stage 'count':
> python process.py numbers.txt > count.txt
Updating lock file 'dvc.lock'
Use `dvc push` to send your updates to remote storage.

DVC 先执行 filter，生成 numbers.txt，然后执行 count，生成最终结果。同时创建 dvc.lock 文件记录当前状态。这个文件类似 Git 的提交记录，保存了每个 Stage 的依赖和输出的哈希值。

增量执行与缓存

DVC 的强大之处在于能跳过未变化的 Stage。修改 process.py 文件，增加一行打印：

import sys
num_lines = 0
with open(sys.argv[1], 'r') as f:
    for line in f:
        num_lines += 1
print('Number of lines:')  # 新增的行
print(num_lines)

再次运行 dvc repro：

$ dvc repro
Stage 'filter' didn't change, skipping
Running stage 'count' with command:
        python process.py numbers.txt > count.txt
Updating lock file 'dvc.lock'

DVC 检测到 filter 的依赖没有变化，直接跳过，只执行受影响的 count Stage。这种增量执行在大规模流水线中节省大量时间。

常用执行选项

dvc repro 提供了丰富的选项控制执行行为：

• --dry：只打印要执行的命令，不实际运行。适合检查流水线状态
• --force：强制重新执行所有 Stage，忽略缓存
• --single-item：只执行指定的 Stage，不递归处理依赖
• --downstream：从指定 Stage 开始，向后执行下游 Stage
• --interactive：每个 Stage 执行前询问确认

比如，只想执行 count Stage，忽略上游变化：

$ dvc repro --single-item count

如果只想执行 filter 之后的 Stage（即 count）：

$ dvc repro --downstream filter

这在调试时很有用，可以快速验证下游 Stage 的逻辑。

处理执行失败

流水线执行中难免遇到错误。默认情况下，一个 Stage 失败会终止整个流程。但有时我们希望继续执行其他独立分支。--keep-going 选项让 DVC 跳过依赖失败 Stage 的其他 Stage，继续执行剩余任务。

$ dvc repro --keep-going

--ignore-errors 更激进，它会忽略所有错误，执行所有能执行的 Stage。这在某些容错场景下有用，但通常不推荐。

缓存机制加速

DVC 的缓存机制是提升效率的核心。它不仅缓存数据文件，还缓存 Stage 的执行结果，这就是运行缓存（run cache）。

运行缓存的工作原理

每次执行 Stage 时，DVC 会计算一个综合哈希值，包括：

• 依赖文件的内容哈希
• 参数值
• 执行命令
• 输出路径

这个哈希值唯一标识一次执行。如果后续执行发现相同的哈希，就直接从缓存恢复输出，不再运行命令。运行缓存存储在 .dvc/cache/runs 目录中。

看个实际例子。修改 params.yaml 中的训练参数：

train:
  n_est: 100  # 从 50 改为 100

执行 dvc repro：

$ dvc repro
Stage 'prepare' didn't change, skipping
Stage 'featurize' didn't change, skipping
Running stage 'train':
> python src/train.py data/features model.pkl
Updating lock file 'dvc.lock'

训练完成后，把参数改回 50，再次执行：

$ dvc repro
Stage 'prepare' didn't change, skipping
Stage 'featurize' didn't change, skipping

这次 train Stage 也没有执行！因为 DVC 发现之前已经用相同的参数执行过，直接从运行缓存恢复了结果。这在调参时特别有用，可以避免重复训练。

控制缓存行为

有时我们不希望使用缓存，比如调试时需要重新运行。--no-run-cache 选项禁用运行缓存：

$ dvc repro --no-run-cache

另一个常见场景是实验性执行，不想污染缓存。--no-commit 选项执行 Stage 但不把输出存入缓存：

$ dvc repro --no-commit

执行后，dvc status 会显示输出文件不在缓存中：

$ dvc status
train:
	changed outs:
		not in cache:       model.pkl

确认结果满意后，用 dvc commit 手动提交到缓存：

$ dvc commit
$ dvc status
Data and pipelines are up to date.

这种方式适合探索性工作，避免生成大量无用的缓存数据。

缓存与 Git 的协同

运行缓存与 Git 提交是独立的。即使切换 Git 分支，运行缓存仍然可用。这意味着在不同实验分支间切换时，如果依赖和参数相同，DVC 可以直接复用缓存的结果，无需重新计算。这种设计让实验管理更加高效。

外部依赖处理

实际项目中，数据不一定都在本地工作区。可能要从外部 URL 下载数据，或者把结果直接写入云存储。DVC 支持外部依赖和输出，让流水线能处理更复杂的场景。

定义外部依赖

外部依赖可以是 HTTP URL、云存储路径等。DVC 会监控这些外部资源的变化。例如，从网络下载数据：

stages:
  download:
    cmd: wget https://example.com/data.zip -O data.zip
    deps:
      - https://example.com/data.zip
    outs:
      - data.zip

每次执行时，DVC 会检查 URL 的内容是否变化（通过 ETag 或 Last-Modified 头）。如果远程文件更新，Stage 会重新执行。

外部输出与缓存控制

有时 Stage 的输出需要直接写入远程存储，不需要本地缓存。比如把处理结果直接上传到 S3：

stages:
  prepare:
    cmd: python sm_prepare.py --bucket mybucket --prefix project-data
    deps:
      - sm_prepare.py
      - https://sagemaker-sample-data.s3.amazonaws.com/data.zip
    outs:
      - s3://mybucket/project-data/input_data:
          cache: false

cache: false 告诉 DVC 不要缓存这个输出。这在处理大规模数据时很有用，避免本地存储成为瓶颈。需要注意的是，标记为 cache: false 的 Stage 会禁用运行缓存，因为 DVC 无法追踪外部存储的内容变化。

外部依赖的注意事项

使用外部依赖时，要确保命令能正确处理网络异常和认证。DVC 不会自动重试失败的下载，这需要在脚本中实现。另外，外部资源的可用性直接影响流水线的可复现性。如果远程数据被删除或修改，历史版本可能无法重现。因此，对于关键数据，建议用 dvc import-url 导入到 DVC 控制下，而不是直接作为外部依赖。

并行执行控制

DVC 的流水线基于 DAG，天然支持并行执行。只要 Stage 之间没有依赖关系，就可以同时运行。

手动并行执行

假设流水线有两个独立分支：

$ dvc dag
+--------+          +--------+
|   A1   |          |   B1   |
+--------+          +--------+
     *                   *
     *                   *
     *                   *
+--------+          +--------+
|   A2   |          |   B2   |
+--------+          +--------+
          *         *
           **     **
             *   *
        +------------+
        |    train   |
        +------------+

A 和 B 两个分支相互独立。可以在两个终端分别执行：

# 终端 1
$ dvc repro A2

# 终端 2
$ dvc repro B2

DVC 会锁定正在执行的 Stage，防止冲突。两个分支执行完成后，再运行最终的 train Stage：

$ dvc repro train

DVC 检测到 A2 和 B2 都已是最新状态，只执行 train。

并行执行的注意事项

并行执行时，要确保 Stage 之间确实没有依赖冲突。如果两个 Stage 写入同一个输出文件，会导致不确定的结果。DVC 的依赖图检查能发现一些明显的冲突，但无法检测逻辑上的竞争条件。

另外，并行执行会增加系统负载。对于计算密集型 Stage，要控制并行数量，避免资源耗尽。可以通过 --jobs 选项限制同时运行的任务数。

自动化并行

对于大规模参数搜索，DVC 提供了实验队列功能。可以把多个参数组合加入队列，然后并行执行：

$ dvc exp run --queue --set-param "train.batch_size=8,16,24"

这会创建三个实验任务。然后用 dvc queue start --jobs 3 启动三个并行 worker 执行。这个功能在超参数调优时特别有用，能充分利用多核 CPU 或集群资源。

总结

流水线构建与执行是 DVC 的核心能力。通过 dvc stage add 或手动编辑 dvc.yaml，我们把数据处理步骤定义为可管理的 Stage。dvc repro 智能地执行流水线，利用缓存机制避免重复工作。外部依赖和并行执行的支持，让流水线能应对真实世界的复杂场景。

掌握这些基础后，下一章将深入参数与指标管理。参数化让流水线更灵活，指标追踪让实验效果可量化。结合本章的构建执行能力，就能搭建完整的机器学习实验框架，实现真正的可复现研究。