03 - Dataset and Model Versioning: DVC vs LakeFS

# Core installation
pip install dvc

# With remote storage support (choose your provider)
pip install dvc[s3]       # Amazon S3
pip install dvc[gs]       # Google Cloud Storage
pip install dvc[azure]    # Azure Blob Storage
pip install dvc[ssh]      # SSH/SFTP
pip install dvc[all]      # All backends

# Verify installation
dvc --version
# DVC 3.x.x

# Initialize Git (if not already done)
git init
git add .
git commit -m "Initial commit"

# Initialize DVC
dvc init

# Structure created by dvc init:
# .dvc/
# ├── config          # DVC configuration (tracked by Git)
# ├── .gitignore      # Excludes cache and tmp
# ├── cache/          # Local artifact cache (NOT tracked by Git)
# └── tmp/

# Commit DVC configuration files
git add .dvc/ .gitignore
git commit -m "chore: initialize DVC"

# Add a dataset to DVC tracking
dvc add data/raw/training_data.csv

# DVC has created:
# - data/raw/training_data.csv.dvc  (pointer, tracked by Git)
# - data/raw/.gitignore             (excludes the original file from Git)

# Contents of the generated .dvc file:
# outs:
# - md5: a1b2c3d4e5f6...
#   size: 524288000
#   path: training_data.csv

# Add the pointer to Git
git add data/raw/training_data.csv.dvc data/raw/.gitignore
git commit -m "feat: add training dataset v1.0"
git tag "dataset-v1.0"

# Add a trained model
dvc add models/churn_model.pkl
git add models/churn_model.pkl.dvc models/.gitignore
git commit -m "feat: add trained churn model v1 (accuracy=0.94)"
git tag "model-v1.0"

# Daily DVC workflow

# 1. Pull updated data from remote
git pull
dvc pull

# 2. Modify dataset or train a new model
python scripts/preprocess.py
python scripts/train.py

# 3. Track new artifacts
dvc add data/processed/features.parquet
dvc add models/churn_model_v2.pkl

# 4. Commit code and pointers
git add .
git commit -m "feat: retrain model with new features (F1=0.92)"

# 5. Push data to remote storage
dvc push

# 6. Push code to Git
git push

# ---- On a colleague's machine ----
git pull          # Downloads code and .dvc files
dvc pull          # Downloads data from remote storage

# ==================== AWS S3 ====================
# Prerequisites: pip install dvc[s3], AWS credentials configured

# Add S3 as the default remote
dvc remote add -d myremote s3://my-ml-bucket/dvc-storage

# For sensitive credentials, use local config (not tracked by Git)
dvc remote modify --local myremote access_key_id $AWS_ACCESS_KEY_ID
dvc remote modify --local myremote secret_access_key $AWS_SECRET_ACCESS_KEY
dvc remote modify myremote region eu-west-1


# ==================== Google Cloud Storage ====================
# Prerequisites: pip install dvc[gs], gcloud auth configured

dvc remote add -d gcsremote gs://my-ml-bucket/dvc-storage

# With Service Account (for production)
dvc remote modify gcsremote credentialpath /path/to/service-account.json


# ==================== Azure Blob Storage ====================
# Prerequisites: pip install dvc[azure]

dvc remote add -d azureremote azure://mycontainer/dvc-storage
dvc remote modify --local azureremote connection_string $AZURE_CONN_STRING


# ==================== Operations ====================
dvc remote list         # Show remote configuration
dvc push                # Push all tracked artifacts
dvc pull                # Pull all artifacts
dvc status --cloud      # Check synchronization status

# DagsHub as free remote storage
dvc remote add origin https://dagshub.com/username/my-ml-project.dvc
dvc remote modify origin --local auth basic
dvc remote modify origin --local user username
dvc remote modify origin --local password $DAGSHUB_TOKEN

dvc push   # Push data
dvc pull   # Pull data

# dvc.yaml - Complete ML pipeline for churn prediction
stages:

  fetch_data:
    cmd: python src/data/fetch_data.py
    deps:
      - src/data/fetch_data.py
    params:
      - data.source_url
      - data.date_range
    outs:
      - data/raw/transactions.parquet
      - data/raw/customers.csv

  preprocess:
    cmd: python src/features/preprocess.py
    deps:
      - src/features/preprocess.py
      - data/raw/transactions.parquet
      - data/raw/customers.csv
    params:
      - features.window_days
      - features.aggregations
    outs:
      - data/processed/features.parquet

  split:
    cmd: python src/data/split.py
    deps:
      - src/data/split.py
      - data/processed/features.parquet
    params:
      - split.train_ratio
      - split.val_ratio
      - split.random_seed
    outs:
      - data/splits/train.parquet
      - data/splits/val.parquet
      - data/splits/test.parquet

  train:
    cmd: python src/models/train.py
    deps:
      - src/models/train.py
      - data/splits/train.parquet
      - data/splits/val.parquet
    params:
      - model.type
      - model.n_estimators
      - model.max_depth
      - model.learning_rate
    outs:
      - models/churn_model.pkl
    metrics:
      - metrics/train_metrics.json:
          cache: false

  evaluate:
    cmd: python src/models/evaluate.py
    deps:
      - src/models/evaluate.py
      - models/churn_model.pkl
      - data/splits/test.parquet
    metrics:
      - metrics/test_metrics.json:
          cache: false

# params.yaml - Centralized configuration
data:
  source_url: "s3://my-data-bucket/raw/2025/"
  date_range: "2024-01-01:2025-01-01"

features:
  window_days: 30
  aggregations: [mean, std, max]

split:
  train_ratio: 0.7
  val_ratio: 0.15
  random_seed: 42

model:
  type: "xgboost"
  n_estimators: 500
  max_depth: 6
  learning_rate: 0.05
  subsample: 0.8

# Pipeline execution and management

dvc repro               # Run pipeline (invalidated stages only)
dvc repro --force       # Force re-run all stages
dvc repro evaluate      # Run up to a specific stage

dvc dag                 # Visualize pipeline DAG

dvc metrics show        # Display current metrics
dvc metrics diff HEAD~1 # Compare with previous commit

dvc params diff main    # Compare params with main branch
dvc diff                # Show data diffs

import dvc.api
import pandas as pd
import pickle
import json


# ==================== dvc.api.open() ====================
# Accesses remote storage directly - no local download needed

with dvc.api.open(
    "data/raw/customers.csv",
    repo="https://github.com/myorg/ml-project",
    rev="dataset-v2.1",
    mode="r"
) as f:
    df = pd.read_csv(f)

print(f"Dataset loaded: {len(df)} rows")

# Read a Parquet file (binary mode)
with dvc.api.open(
    "data/processed/features.parquet",
    rev="model-v3.0",
    mode="rb"
) as f:
    features_df = pd.read_parquet(f)


# ==================== dvc.api.get_url() ====================
# Gets the direct URL in remote storage for a specific version

url = dvc.api.get_url(
    "models/churn_model.pkl",
    repo="https://github.com/myorg/ml-project",
    rev="model-v2.5"
)
print(f"Model URL: {url}")
# Output: s3://my-ml-bucket/dvc-storage/ab/cd1234...


# ==================== Versioned Model Loading ====================
def load_model(version: str):
    """Load a model from a specific DVC version."""
    with dvc.api.open("models/churn_model.pkl", rev=version, mode="rb") as f:
        model = pickle.load(f)
    with dvc.api.open("models/feature_importance.json", rev=version, mode="r") as f:
        metadata = json.load(f)
    return model, metadata


model, metadata = load_model("model-v3.0")
print(f"Model loaded. Features: {metadata['n_features']}")

# Feature branch workflow

# 1. Create experiment branch
git checkout -b experiment/xgboost-v2

# 2. Update params.yaml (e.g. learning_rate: 0.01)

# 3. Run pipeline
dvc repro

# 4. Check and compare metrics
dvc metrics show
dvc metrics diff main
# accuracy: 0.9301 -> 0.9423  (+0.0122)

# 5. Commit and push
git add dvc.lock params.yaml metrics/
git commit -m "experiment: XGBoost v2 - accuracy +1.2% (0.9423)"
dvc push
git push origin experiment/xgboost-v2

# 6. After PR merge, tag the production model
git checkout main
git tag -a "model-v2.0" -m "XGBoost v2: accuracy=0.9423"
git push --tags


# ==================== Rollback in Production ====================
# Model v2.0 has a bug - roll back to v1.0

git checkout model-v1.0   # Go back to that commit
dvc checkout              # Restore data and models to that state
python src/models/evaluate.py  # Verify restored model

import mlflow
import mlflow.sklearn
import dvc.api
import subprocess
import pandas as pd
from sklearn.metrics import accuracy_score, f1_score, roc_auc_score
from xgboost import XGBClassifier


def get_dvc_metadata() -> dict:
    """Collect DVC metadata for the current MLflow run."""
    git_rev = subprocess.run(
        ["git", "rev-parse", "HEAD"], capture_output=True, text=True
    ).stdout.strip()

    data_url = dvc.api.get_url("data/splits/train.parquet")

    return {
        "dvc.git_rev": git_rev,
        "dvc.data_url": data_url,
        "dvc.data_version": "v2.1",
    }


def train_with_tracking(params: dict) -> None:
    """Full training run with DVC + MLflow tracking."""
    mlflow.set_experiment("churn-prediction")

    with mlflow.start_run(run_name="xgboost-dvc-integrated") as run:
        # Link MLflow run to data version
        mlflow.log_params(get_dvc_metadata())
        mlflow.log_params(params)

        # Load versioned data
        train_df = pd.read_parquet("data/splits/train.parquet")
        test_df = pd.read_parquet("data/splits/test.parquet")

        X_train = train_df.drop("churn", axis=1)
        y_train = train_df["churn"]
        X_test = test_df.drop("churn", axis=1)
        y_test = test_df["churn"]

        model = XGBClassifier(**params)
        model.fit(X_train, y_train)

        y_pred = model.predict(X_test)
        y_prob = model.predict_proba(X_test)[:, 1]

        metrics = {
            "accuracy": accuracy_score(y_test, y_pred),
            "f1_score": f1_score(y_test, y_pred),
            "auc_roc": roc_auc_score(y_test, y_prob),
        }

        mlflow.log_metrics(metrics)
        mlflow.sklearn.log_model(model, "model")
        print(f"Run {run.info.run_id} | Accuracy: {metrics['accuracy']:.4f}")

# lakeFS: Python SDK workflow example
from lakefs_sdk import Client

client = Client(
    host="https://lakefs.mycompany.com",
    username="access_key",
    password="secret_key"
)

repo = client.repositories.create_repository(
    "churn-data-lake",
    storage_namespace="s3://my-data-lake/churn/",
    default_branch="main"
)

# Create experiment branch (zero-copy - milliseconds at any scale)
experiment_branch = repo.branch("experiment/new-features").create(
    source_reference="main"
)

# Upload new data without impacting main branch
experiment_branch.object("data/features_v2.parquet").upload(
    content=new_features_df.to_parquet()
)

experiment_branch.commit(
    message="feat: add recency features",
    metadata={"model_version": "v3.0"}
)

# Merge after successful validation
repo.branch("main").merge(source_ref="experiment/new-features")

# Instant rollback if needed
repo.branch("main").revert(reference="main~1")

ml-project/
├── .dvc/
│   ├── config              # Remote config (no secrets)
│   └── .gitignore
├── data/
│   ├── raw/                # .dvc pointers + .gitignore
│   ├── processed/          # .dvc pointers
│   └── splits/             # .dvc pointers
├── models/                 # .dvc pointers + .gitignore
├── metrics/                # JSON files tracked by Git
├── reports/                # PNG reports (cache: false)
├── src/
│   ├── data/
│   ├── features/
│   └── models/
├── dvc.yaml                # Pipeline definition
├── dvc.lock                # Pipeline state (ALWAYS commit to Git)
├── params.yaml             # Hyperparameters
└── requirements.txt

# .github/workflows/ml-pipeline.yml
name: ML Pipeline Validation

on:
  pull_request:
    branches: [main]
    paths: ['src/**', 'params.yaml', 'dvc.yaml']

jobs:
  run-pipeline:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - run: pip install dvc[s3] -r requirements.txt

      - name: Configure DVC remote
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
        run: |
          dvc remote modify --local myremote \
            access_key_id $AWS_ACCESS_KEY_ID
          dvc remote modify --local myremote \
            secret_access_key $AWS_SECRET_ACCESS_KEY

      - run: dvc pull data/splits/
      - run: dvc repro

      - name: Check metrics regression
        run: |
          dvc metrics diff main --md >> $GITHUB_STEP_SUMMARY
          python scripts/check_metrics.py --min-accuracy 0.90

      - run: dvc push