Data Pipelines for AI
Data pipelines for AI ensure clean, versioned, and feature-engineered data flows from sources to model training and inference.
Feature Engineering Pipeline
python
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Callable
@dataclass
class Feature:
name: str
transform: Callable
description: str
class FeaturePipeline:
def __init__(self):
self.features: List[Feature] = []
def add_feature(self, name: str, transform: Callable, description: str = ""):
self.features.append(Feature(name, transform, description))
def transform(self, df: pd.DataFrame) -> pd.DataFrame:
result = df.copy()
for feature in self.features:
result[feature.name] = feature.transform(result)
return result
def get_feature_names(self) -> List[str]:
return [f.name for f in self.features]
# Usage
pipeline = FeaturePipeline()
pipeline.add_feature("revenue_per_unit", lambda df: df["revenue"] / df["units_sold"])
pipeline.add_feature("profit_margin", lambda df: (df["revenue"] - df["cost"]) / df["revenue"])
pipeline.add_feature("month", lambda df: pd.to_datetime(df["date"]).dt.month)
Data Version Control
python
import hashlib
import json
from datetime import datetime
class DataVersionControl:
def __init__(self, storage_path: str = "data_versions"):
self.storage_path = storage_path
self.versions = {}
def commit(self, dataset_name: str, df: pd.DataFrame, message: str) -> str:
content_hash = hashlib.sha256(df.to_csv().encode()).hexdigest()[:12]
version = {
"hash": content_hash,
"timestamp": datetime.now().isoformat(),
"message": message,
"rows": len(df),
"columns": list(df.columns),
"schema": df.dtypes.astype(str).to_dict()
}
if dataset_name not in self.versions:
self.versions[dataset_name] = []
self.versions[dataset_name].append(version)
df.to_parquet(f"{self.storage_path}/{dataset_name}_{content_hash}.parquet")
return content_hash
def checkout(self, dataset_name: str, version_hash: str) -> pd.DataFrame:
return pd.read_parquet(f"{self.storage_path}/{dataset_name}_{version_hash}.parquet")
def log(self, dataset_name: str) -> list:
return self.versions.get(dataset_name, [])
# Usage
dvc = DataVersionControl()
hash1 = dvc.commit("training_data", train_df, "Initial dataset")
hash2 = dvc.commit("training_data", updated_df, "Added new features")
train_df = dvc.checkout("training_data", hash1)
Data Quality Monitor
python
class DataQualityMonitor:
def __init__(self):
self.rules = []
self.metrics_history = []
def add_rule(self, name: str, check_fn, severity: str = "error"):
self.rules.append({"name": name, "check": check_fn, "severity": severity})
def check_quality(self, df: pd.DataFrame) -> dict:
results = {"passed": [], "failed": [], "warnings": []}
for rule in self.rules:
try:
passed = rule["check"](df)
if passed:
results["passed"].append(rule["name"])
else:
entry = {"name": rule["name"], "severity": rule["severity"]}
if rule["severity"] == "error":
results["failed"].append(entry)
else:
results["warnings"].append(entry)
except Exception as e:
results["failed"].append({"name": rule["name"], "error": str(e)})
results["overall"] = len(results["failed"]) == 0
results["score"] = len(results["passed"]) / len(self.rules) if self.rules else 1.0
self.metrics_history.append(results)
return results
def add_default_rules(self):
self.add_rule("no_nulls", lambda df: df.isnull().sum().sum() == 0, "warning")
self.add_rule("min_rows", lambda df: len(df) > 100, "error")
self.add_rule("no_duplicates", lambda df: df.duplicated().sum() == 0, "warning")
# Usage
monitor = DataQualityMonitor()
monitor.add_default_rules()
quality = monitor.check_quality(df)
Key Takeaways
- Feature pipelines automate feature creation and transformation
- Data versioning tracks dataset changes for reproducibility
- Quality monitoring catches data issues before training
- Feature stores serve features consistently for training and inference
- Lineage tracking documents data origins and transformations