Expert patterns for reproducible data management with DVC, Delta Lake, and data lineage tracking.
Pattern 1: Git + DVC Workflow
The core pattern: code in Git, data pointers in Git, data bytes in remote storage.
# Initialize DVC in existing Git repo
git init
dvc init
git add .dvc/
git commit -m "chore: initialize DVC"
dvc remote add -d s3remote s3://my-ml-bucket/dvc-cache dvc remote modify s3remote region us-east-1 git add .dvc/config git commit -m "chore: configure DVC S3 remote"
Track a datasetdvc add data/raw/train.parquet git add data/raw/train.parquet.dvc data/raw/.gitignore git commit -m "data: add raw training dataset v1" dvc push # uploads actual file to S3
Later: recover exact data for any commitgit checkout abc1234 dvc checkout # downloads the data version from that commit
outs:
- md5: a1b2c3d4e5f6... size: 1073741824 path: train.parquet
## Pattern 2: DVC Pipeline for Reproducible ML
Define the full pipeline in `dvc.yaml` so any stage can be reproduced.
```yaml
# dvc.yaml
stages:
preprocess:
cmd: python src/preprocess.py --input data/raw/train.parquet --output data/processed/
deps:
- src/preprocess.py
- data/raw/train.parquet
params:
- params.yaml:
- preprocess.max_seq_len
- preprocess.lowercase
outs:
- data/processed/train.parquet
- data/processed/val.parquet
train:
cmd: python src/train.py --data data/processed/ --model models/
deps:
- src/train.py
- data/processed/train.parquet
- data/processed/val.parquet
params:
- params.yaml:
- train.learning_rate
- train.batch_size
- train.epochs
outs:
- models/model.pt
metrics:
- metrics/eval.json:
cache: false
# params.yaml
preprocess:
max_seq_len: 512
lowercase: true
train:
learning_rate: 3.0e-4
batch_size: 32
epochs: 10
# Reproduce only changed stages
dvc repro
# See what changed
dvc params diff HEAD~1
dvc metrics diff HEAD~1
# Visualize pipeline
dvc dag
Pattern 3: DVC Experiments for Hyperparameter Search
Run experiments without creating Git commits for each trial.
# Run experiment with modified param
dvc exp run --set-param train.learning_rate=1e-4
# Run a grid of experiments
dvc exp run --set-param train.learning_rate=1e-3,1e-4,3e-4 \
--set-param train.batch_size=16,32
# Show all experiments
dvc exp show
# Promote best experiment to a branch
dvc exp branch exp-lr-1e4 feature/best-lr
# Clean up failed experiments
dvc exp gc --workspace
Pattern 4: Delta Lake Time Travel
Access any historical version of a dataset by timestamp or version number.
from delta import DeltaTable
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.config("spark.jars.packages", "io.delta:delta-core_2.12:2.4.0") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.getOrCreate()
delta_path = "s3://my-bucket/features/user_features"
# Read latest version
df_current = spark.read.format("delta").load(delta_path)
# Read specific version
df_v3 = spark.read.format("delta") \
.option("versionAsOf", 3) \
.load(delta_path)
# Read by timestamp (training data as of model release date)
df_at_release = spark.read.format("delta") \
.option("timestampAsOf", "2024-06-01 00:00:00") \
.load(delta_path)
# Inspect history
dt = DeltaTable.forPath(spark, delta_path)
history = dt.history()
history.select("version", "timestamp", "operation", "operationParameters").show()
# Restore to previous version if bad data was written
dt.restoreToVersion(3)
Pattern 5: OpenLineage Event Emission
Emit lineage events from custom data pipelines to Marquez or any OpenLineage-compatible backend.
from openlineage.client import OpenLineageClient
from openlineage.client.run import (
RunEvent, RunState, Run, Job, Dataset, InputDataset, OutputDataset
)
from openlineage.client.facet import SchemaDatasetFacet, SchemaField
import uuid
from datetime import datetime
client = OpenLineageClient.from_environment() # reads OPENLINEAGE_URL env var
def emit_lineage(
job_name: str,
input_datasets: list[str],
output_datasets: list[str],
namespace: str = "default"
):
run_id = str(uuid.uuid4())
inputs = [InputDataset(namespace=namespace, name=ds) for ds in input_datasets]
outputs = [OutputDataset(namespace=namespace, name=ds) for ds in output_datasets]
# Start event
client.emit(RunEvent(
eventType=RunState.START,
eventTime=datetime.utcnow().isoformat() + "Z",
run=Run(runId=run_id),
job=Job(namespace=namespace, name=job_name),
inputs=inputs,
outputs=outputs,
))
return run_id
def complete_lineage(run_id: str, job_name: str, namespace: str = "default"):
client.emit(RunEvent(
eventType=RunState.COMPLETE,
eventTime=datetime.utcnow().isoformat() + "Z",
run=Run(runId=run_id),
job=Job(namespace=namespace, name=job_name),
))
# Usage
run_id = emit_lineage(
job_name="preprocess_features",
input_datasets=["raw.user_events"],
output_datasets=["features.user_features_v2"]
)
# ... do actual work ...
complete_lineage(run_id, "preprocess_features")
Pattern 6: Versioned Dataset Splits with DVC
Save exact split indices alongside data to guarantee reproducibility.
import numpy as np
import pandas as pd
from sklearn.model_selection import StratifiedKFold
import json
def create_versioned_splits(
df: pd.DataFrame,
label_col: str,
n_splits: int = 5,
val_fold: int = 0,
random_state: int = 42
) -> dict:
"""
Create deterministic, stratified train/val/test splits.
Returns indices (not copies of data) for storage efficiency.
"""
# Hold out test set first
from sklearn.model_selection import train_test_split
train_val_idx, test_idx = train_test_split(
np.arange(len(df)),
test_size=0.1,
stratify=df[label_col].values,
random_state=random_state
)
# K-fold on train+val
skf = StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=random_state)
folds = list(skf.split(train_val_idx, df[label_col].values[train_val_idx]))
train_fold_idx, val_fold_idx = folds[val_fold]
train_idx = train_val_idx[train_fold_idx]
val_idx = train_val_idx[val_fold_idx]
split_spec = {
"random_state": random_state,
"val_fold": val_fold,
"n_splits": n_splits,
"train_indices": train_idx.tolist(),
"val_indices": val_idx.tolist(),
"test_indices": test_idx.tolist(),
"label_distribution": {
"train": df[label_col].values[train_idx].tolist(),
"val": df[label_col].values[val_idx].tolist(),
}
}
# Save and track with DVC
with open("data/splits/split_v1.json", "w") as f:
json.dump(split_spec, f)
# then: dvc add data/splits/split_v1.json && git commit
return split_spec
Anti-Patterns
Anti-Pattern 1: Storing Data in Git
Binary files in Git bloat the repo permanently. Even after deletion they remain in history. Git LFS is a partial solution but still has limits. Use DVC or Delta Lake for data files of any significant size.
Anti-Pattern 2: Using Filenames as Versions
train_v2_final_FINAL.csv is not version control. There is no diff, no author, no timestamp, no audit trail. Every dataset must have a hash-based identifier.
Anti-Pattern 3: Mutating the Training Dataset
Overwriting a training dataset that a model checkpoint was trained on makes the model unauditable. Training datasets must be append-only or versioned. Use Delta Lake ACID transactions or DVC tags to freeze versions.
Anti-Pattern 4: Ignoring Split Reproducibility
Train/val/test split is part of the experiment. If the split is random and untracked, two "identical" experiments may have seen different validation examples. Always save split indices.
Anti-Pattern 5: No Lineage Between Code and Data
Without lineage, you cannot answer: "What version of data was this model trained on?" Link every model artifact to the exact Git commit + DVC lock file that produced it.