# Add a large dataset to DVC tracking
dvc add data/raw/survey_2023.csv
# This creates:# - data/raw/survey_2023.csv.dvc (commit this to Git)# - .gitignore updated to ignore the actual CSV
git add data/raw/survey_2023.csv.dvc data/raw/.gitignore
git commit -m "Add survey 2023 dataset to DVC"# Push data to remote
dvc push
# Later: reproduce from another machine
git clone https://github.com/org/my-research-project
cd my-research-project
dvc pull # downloads data from remote
# Run the full pipeline
dvc repro
# View metrics
dvc metrics show
# Visualize the DAG
dvc dag
Step 3: Experiment Tracking and Comparison
# File: scripts/run_experiments.py"""Run multiple experiments varying hyperparameters."""import subprocess
import json
import yaml
from pathlib import Path
import pandas as pd
defrun_dvc_experiment(params_override, exp_name=None):
"""Run a DVC experiment with given parameter overrides.
Args:
params_override: dict of param_path → value (e.g., "train.n_estimators" → 200)
exp_name: optional experiment name
Returns:
dict with command result
"""
cmd = ["dvc", "exp", "run"]
if exp_name:
cmd.extend(["--name", exp_name])
for param_path, value in params_override.items():
cmd.extend(["--set-param", f"{param_path}={value}"])
print(f"Running: {' '.join(cmd)}")
result = subprocess.run(cmd, capture_output=True, text=True)
return {
"returncode": result.returncode,
"stdout": result.stdout,
"stderr": result.stderr,
"success": result.returncode == 0,
}
defget_experiments_table():
"""Retrieve experiment comparison table from DVC."""
result = subprocess.run(
["dvc", "exp", "show", "--csv"],
capture_output=True, text=True
)
if result.returncode == 0:
import io
return pd.read_csv(io.StringIO(result.stdout))
return pd.DataFrame()
# Define experiment grid
experiments = [
{
"name": "rf-100-depth5",
"params": {
"train.model_type": "random_forest",
"train.n_estimators": 100,
"train.max_depth": 5,
}
},
{
"name": "rf-200-depth8",
"params": {
"train.model_type": "random_forest",
"train.n_estimators": 200,
"train.max_depth": 8,
}
},
{
"name": "gb-100-depth4",
"params": {
"train.model_type": "gradient_boosting",
"train.n_estimators": 100,
"train.max_depth": 4,
}
},
]
print("=== DVC Experiment Grid ===")
for exp in experiments:
print(f"\nExperiment: {exp['name']}")
for k, v in exp['params'].items():
print(f" {k} = {v}")
# In a real workflow, uncomment:# for exp in experiments:# result = run_dvc_experiment(exp["params"], exp_name=exp["name"])# print(f" {'OK' if result['success'] else 'FAILED'}: {exp['name']}")# Load results from metrics files directly (as DVC substitute)
metrics_files = list(Path("metrics").glob("*.json"))
results = []
for f in metrics_files:
withopen(f) as fp:
data = json.load(fp)
ifisinstance(data, dict) and"test_rmse"in data:
data["file"] = str(f)
results.append(data)
if results:
df_results = pd.DataFrame(results)
print("\n=== Current Metrics ===")
print(df_results.to_string(index=False))
Advanced Usage
DVC Data Registry Pattern
# Register shared datasets accessible to all team projects# In a central "data-registry" repo
dvc add data/gold/benchmark_dataset_v2.parquet
git commit -am "Add benchmark dataset v2"
git tag -a "benchmark-v2.0" -m "Benchmark dataset version 2"
dvc push
# In a downstream project
dvc import git@github.com:org/data-registry.git \
data/gold/benchmark_dataset_v2.parquet \
-o data/benchmark.parquet
# Get updates when registry is updated
dvc update benchmark.parquet.dvc
Parameterized Pipeline via Python
# File: scripts/parametric_run.py"""Generate DVC pipeline configurations programmatically."""import yaml
from pathlib import Path
from itertools import product
defgenerate_dvc_yaml(model_types, n_estimators_list):
"""Generate dvc.yaml with multiple training configurations."""
stages = {}
for model, n_est in product(model_types, n_estimators_list):
stage_name = f"train_{model}_{n_est}"
stages[stage_name] = {
"cmd": f"python src/train.py --model {model} --n-estimators {n_est}",
"deps": ["data/processed/X_train.npy", "src/train.py"],
"outs": [f"models/{model}_{n_est}/model.pkl"],
"metrics": [{f"metrics/{model}_{n_est}_metrics.json": {"cache": False}}],
"params": ["params.yaml:train.random_seed"],
}
dvc_config = {"stages": stages}
withopen("dvc_grid.yaml", "w") as f:
yaml.dump(dvc_config, f, default_flow_style=False)
print(f"Generated dvc_grid.yaml with {len(stages)} stages")
return stages
stages = generate_dvc_yaml(
model_types=["random_forest", "gradient_boosting"],
n_estimators_list=[50, 100, 200]
)
# Run specific stage# dvc repro -f dvc_grid.yaml dvc_grid:train_random_forest_100
Integration with MLflow
# File: src/train_with_mlflow.py"""Train model with MLflow tracking alongside DVC."""import numpy as np
import yaml
import pickle
import json
import os
from pathlib import Path
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import r2_score, mean_squared_error
MLFLOW_TRACKING_URI = os.getenv("MLFLOW_TRACKING_URI", "http://localhost:5000")
deftrain_with_tracking():
withopen("params.yaml") as f:
params = yaml.safe_load(f)
X_train = np.load("data/processed/X_train.npy")
y_train = np.load("data/processed/y_train.npy")
X_test = np.load("data/processed/X_test.npy")
y_test = np.load("data/processed/y_test.npy")
cfg = params["train"]
model = RandomForestRegressor(
n_estimators=cfg.get("n_estimators", 100),
max_depth=cfg.get("max_depth", 5),
random_state=cfg.get("random_seed", 42)
)
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
metrics = {
"test_rmse": float(np.sqrt(mean_squared_error(y_test, y_pred))),
"test_r2": float(r2_score(y_test, y_pred)),
}
try:
import mlflow
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
with mlflow.start_run(run_name="dvc_experiment"):
mlflow.log_params(cfg)
mlflow.log_metrics(metrics)
mlflow.sklearn.log_model(model, "model")
print(f"Logged to MLflow at {MLFLOW_TRACKING_URI}")
except ImportError:
print("MLflow not installed — logging to JSON only")
except Exception as e:
print(f"MLflow logging failed ({e}) — continuing with JSON")
Path("metrics").mkdir(exist_ok=True)
withopen("metrics/test_metrics.json", "w") as f:
json.dump(metrics, f, indent=2)
Path("models").mkdir(exist_ok=True)
withopen("models/model.pkl", "wb") as f:
pickle.dump(model, f)
print(f"Test RMSE: {metrics['test_rmse']:.4f}, R²: {metrics['test_r2']:.4f}")
if __name__ == "__main__":
train_with_tracking()
Troubleshooting
Problem
Cause
Fix
dvc: command not found
DVC not installed
pip install dvc; ensure venv is activated
ERROR: Git is not initialized
Running dvc init outside git repo
git init first, then dvc init
Remote push fails (S3)
Missing AWS credentials
Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY env vars
Pipeline doesn't re-run
DVC thinks stage is up-to-date
dvc repro --force to override cache
Stage skipped despite code change
Script not in deps list
Add script file to deps: in dvc.yaml
dvc.lock conflict (git merge)
Parallel pipeline runs
Resolve manually; keep the run with better metrics
