DVC技术问题：如何循环复用流水线处理多源文件？

方法1：DVC原生foreach语法（推荐，DVC 2.0+） #

DVC内置的foreach可以直接遍历文件/参数列表，自动生成复用同一逻辑的子stage，同时保留细粒度的依赖跟踪。

1. 定义输入文件列表
   可以在params.yaml中显式列出所有源文件：

   input_files:
     - data/src/file1.txt
     - data/src/file2.txt
     ...
     - data/src/file100.txt
   

   或者用通配符自动匹配（文件名有规律时更高效）：

   vars:
     input_files: !path data/src/*.txt
   

2. 编写循环流水线
   在dvc.yaml中用foreach遍历输入文件，动态生成每个文件的处理逻辑：

   stages:
     process_files:
       foreach: ${input_files}
       do:
         cmd: python process.py ${item} data/out/${item.stem}.processed.txt
         deps:
           - ${item}
           - process.py
         outs:
           - data/out/${item.stem}.processed.txt
   

   • ${item}会依次替换为每个输入文件路径
   • ${item.stem}提取文件名（不含扩展名），用来生成对应的输出文件名

3. 执行流水线
   运行dvc repro，DVC会自动为每个输入文件创建独立子stage，仅处理发生变化的文件。

方法2：脚本动态生成DVC Stages #

如果需要更复杂的文件名映射或过滤逻辑，可以用脚本批量生成dvc.yaml内容。

示例Python脚本generate_stages.py：

from pathlib import Path
import yaml

src_dir = Path("data/src")
out_dir = Path("data/out")
out_dir.mkdir(exist_ok=True)

stage_config = {"stages": {}}

for src_path in src_dir.glob("*.txt"):
    stage_name = f"process_{src_path.stem}"
    out_path = out_dir / f"{src_path.stem}.processed.txt"
    
    stage_config["stages"][stage_name] = {
        "cmd": f"python process.py {src_path} {out_path}",
        "deps": [str(src_path), "process.py"],
        "outs": [str(out_path)]
    }

with open("dvc.yaml", "w") as f:
    yaml.dump(stage_config, f)

运行脚本生成dvc.yaml后，执行dvc repro即可批量处理所有文件。

方法3：单Stage内的循环命令（适合简单场景） #

如果不需要跟踪单个文件的依赖变化，可以在单个stage中用shell循环处理所有文件，但这种方式DVC会把整个目录作为依赖/输出：

stages:
  process_all:
    cmd: |
      for src_file in data/src/*.txt; do
        out_file="data/out/$(basename "$src_file" .txt).processed.txt"
        python process.py "$src_file" "$out_file"
      done
    deps:
      - data/src/
      - process.py
    outs:
      - data/out/

注意：任何一个源文件变化都会触发整个stage重新运行。

内容的提问来源于stack exchange，提问作者GlaceCelery