How do I decide between batch and streaming pipelines?

Use the feature freshness requirement as your guide. If the model needs data updated in under 5 minutes, you need a streaming pipeline. If daily or hourly updates are sufficient, batch is simpler, cheaper, and easier to debug. Most AI products use a hybrid: real-time features (session data, current context) via streaming, and historical aggregates (30-day purchase count, lifetime value) via batch.

What is feature drift and how do I detect it?

Feature drift means the statistical distribution of your input features has changed compared to what the model was trained on. Detect it by computing Population Stability Index (PSI) or Kolmogorov-Smirnov tests daily against your training data distribution. A PSI above 0.2 warrants investigation. Common causes: upstream schema changes, seasonality, new user segments, or data collection bugs.

How much data do I need to start training?

It depends on the model type. For fine-tuning LLMs, 500-2,000 high-quality labeled examples can produce meaningful improvements. For traditional ML classifiers, 10,000-100,000 labeled examples is a reasonable starting point. For recommendation systems, you typically need 3-6 months of behavioral data. Start with less data, validate the approach works, then invest in scaling data collection.

Should I use a managed feature store or build my own?

For teams with fewer than 5 ML engineers and fewer than 50 features, a simple solution (Redis for online, S3/Parquet for offline) works fine. Managed feature stores (Feast, Tecton, Hopsworks) become worthwhile when you have 50+ features, multiple models sharing features, or strict freshness SLAs that require automated materialization. The operational overhead of running a feature store is real. Do not adopt one prematurely.

How do I handle data quality issues that only show up in production?

Build a feedback loop. Log model predictions alongside the features used. When users report bad predictions, trace back to the input features. Implement automated anomaly detection on feature values at serving time (not just in the batch pipeline). Add circuit breakers that fall back to default predictions when feature quality drops below thresholds.

AI Data Pipeline Template

What This Template Does

Most AI product failures are data failures. The model architecture gets 80% of the attention, but data quality, pipeline reliability, and feature freshness determine whether the product actually works. A recommendation engine trained on stale data recommends products that are out of stock. A fraud detection model that misses a schema change in the payments table starts flagging legitimate transactions. A chatbot whose knowledge base index is two weeks behind gives users outdated answers.

This template provides a structured spec for designing the data infrastructure that feeds your AI product. It covers data sources, collection mechanisms, preprocessing logic, feature engineering, training/serving pipelines, data quality monitoring, and governance. The AI PM Handbook covers the broader AI product lifecycle, and the AI Product PRD Template provides the overall requirements framework. Use the AI ROI Calculator to model how pipeline investments affect total AI project cost.

Direct Answer

An AI Data Pipeline Template is a product spec for the data infrastructure behind an AI feature or product. It defines data sources, collection and preprocessing steps, feature engineering, training and serving pipeline architecture, freshness requirements, quality monitoring, and data governance policies. Use it to ensure your AI product has reliable, high-quality data from day one.

Template Structure

1. Pipeline Overview and Data Requirements

Purpose: Define what data your AI product needs, where it comes from, and how fresh it must be.

## Pipeline Overview

**AI Product/Feature**: [Name]
**Pipeline Owner**: [Data engineer or ML engineer responsible]
**Data Science Lead**: [Person defining feature requirements]
**Environment**: [Cloud provider / On-premises / Hybrid]

### Data Requirements Summary
| Data Type | Source | Volume | Freshness Requirement | Criticality |
|-----------|--------|--------|----------------------|-------------|
| User behavior events | Event stream | [events/day] | Real-time (< 1 min) | High |
| Product catalog | Database | [records] | Near real-time (< 1 hr) | High |
| User profiles | Database | [records] | Daily | Medium |
| Transaction history | Data warehouse | [records/month] | Daily | High |
| External data (market, weather) | API | [calls/day] | Hourly | Low |
| Training labels | Annotation platform | [labels/week] | Weekly | High |

### Data Flow Diagram
[Describe or reference the high-level data flow: sources → ingestion → storage → processing → feature store → model training/serving]

2. Data Collection and Ingestion

Purpose: Define how raw data enters the pipeline, including event schemas, API integrations, and batch imports.

## Data Collection

### Event Stream Specification
**Event Bus**: [Kafka / Kinesis / Pub/Sub / EventBridge]
**Schema Registry**: [Confluent / AWS Glue / Custom]
**Event Format**: [Avro / Protobuf / JSON]

| Event Type | Schema Version | Avg Size | Rate | Retention |
|-----------|---------------|----------|------|-----------|
| page_view | v3.1 | [bytes] | [events/sec] | 90 days |
| purchase | v2.4 | [bytes] | [events/sec] | 365 days |
| search_query | v1.2 | [bytes] | [events/sec] | 180 days |
| user_feedback | v1.0 | [bytes] | [events/sec] | 365 days |

### Batch Data Sources
| Source | Extraction Method | Schedule | Format | Volume |
|--------|------------------|----------|--------|--------|
| Product DB | CDC / Full dump | Hourly | Parquet | [GB] |
| CRM | API pull | Daily | JSON | [GB] |
| External API | Scheduled fetch | Hourly | JSON | [MB] |
| Annotation exports | Manual upload | Weekly | CSV | [MB] |

### Ingestion Reliability
- [ ] Dead letter queue for failed events
- [ ] Schema validation at ingestion point
- [ ] Duplicate detection and deduplication
- [ ] Backfill capability for historical data
- [ ] Monitoring for ingestion lag and data loss

3. Data Preprocessing and Transformation

Purpose: Define the cleaning, normalization, and transformation steps that turn raw data into model-ready inputs.

## Preprocessing Pipeline

### Data Quality Rules
| Rule | Applied To | Action on Failure |
|------|-----------|------------------|
| Null check | Required fields | Drop record + alert |
| Range validation | Numeric fields | Cap to valid range |
| Format validation | Dates, emails, IDs | Drop record + alert |
| Freshness check | All batch sources | Block pipeline + alert |
| Duplicate detection | All sources | Keep latest, drop duplicates |
| Outlier detection | Numeric features | Flag for review |

### Transformation Steps
| Step | Input | Output | Logic | Dependencies |
|------|-------|--------|-------|-------------|
| User session assembly | Raw events | Session objects | Group by user + 30min timeout | Event stream |
| Text normalization | Raw text fields | Cleaned text | Lowercase, strip HTML, normalize unicode | None |
| Category encoding | Categorical fields | Numeric vectors | One-hot or embedding lookup | Category mapping table |
| Timestamp features | Raw timestamps | Time-based features | Hour of day, day of week, recency | None |
| Aggregation | Transaction history | User-level aggregates | Sum, count, avg over 7/30/90 day windows | Transaction data |

### Processing Infrastructure
**Batch Processing**: [Spark / dbt / Airflow / Dagster]
**Stream Processing**: [Flink / Spark Streaming / Kafka Streams]
**Orchestration**: [Airflow / Dagster / Prefect / Step Functions]
**Schedule**: [Cron expression or trigger description]
**Compute**: [Instance types, cluster size, auto-scaling rules]

4. Feature Engineering and Feature Store

Purpose: Define the features your model needs, how they are computed, and how they are served.

## Feature Engineering

### Feature Catalog
| Feature Name | Type | Source | Computation | Freshness | Used In |
|-------------|------|--------|-------------|-----------|---------|
| user_purchase_count_30d | Numeric | Transactions | COUNT over 30d window | Daily | Rec model |
| user_avg_order_value | Numeric | Transactions | AVG over all time | Daily | Rec model |
| product_popularity_score | Numeric | Events | Weighted view/purchase ratio | Hourly | Ranking |
| user_text_embedding | Vector (768d) | Profile + reviews | Sentence transformer | Weekly | Personalization |
| session_intent_score | Numeric | Events | Real-time click pattern classifier | Real-time | Ranking |

### Feature Store Configuration
**Feature Store**: [Feast / Tecton / Hopsworks / SageMaker Feature Store / Custom]
**Offline Store**: [S3/GCS + Parquet / Delta Lake / BigQuery]
**Online Store**: [Redis / DynamoDB / Bigtable]
**Feature Freshness SLA**: [Max staleness per feature tier]

### Feature Freshness Tiers
| Tier | Freshness | Computation | Storage | Example Features |
|------|-----------|-------------|---------|-----------------|
| Real-time | < 1 min | Stream processing | Online store | Session intent, current cart |
| Near real-time | < 1 hr | Micro-batch | Online store | Trending products, user activity |
| Daily | < 24 hr | Batch pipeline | Online + offline | Purchase aggregates, profile features |
| Weekly | < 7 days | Batch pipeline | Offline only | Embeddings, segmentation labels |

### Feature Validation
- [ ] Feature distributions monitored for drift (KL divergence, PSI)
- [ ] Feature coverage tracked (% of entities with non-null values)
- [ ] Feature importance validated against model performance
- [ ] Unused features pruned quarterly
- [ ] Feature lineage documented (source → transformation → feature)

5. Training and Serving Pipelines

Purpose: Define how models are trained, validated, and served using pipeline data.

## Training Pipeline

### Training Data Specification
**Training Set Size**: [Records or examples]
**Validation Set Size**: [Records]
**Test Set Size**: [Records]
**Split Strategy**: [Time-based / Random / Stratified]
**Label Source**: [Implicit feedback / Annotations / Business rules]
**Label Delay**: [Time between event and label availability]

### Training Schedule
| Model | Training Frequency | Training Duration | Trigger |
|-------|-------------------|-------------------|---------|
| Recommendation model | Weekly | [hours] | Scheduled (Sunday 2am) |
| Fraud classifier | Daily | [hours] | Scheduled (midnight) |
| Search ranking | Biweekly | [hours] | Manual trigger after feature changes |

### Model Validation Gates
- [ ] Accuracy above threshold on test set
- [ ] No regression vs. current production model
- [ ] Fairness metrics within tolerance
- [ ] Latency within SLA on benchmark inputs
- [ ] Feature coverage above minimum threshold

## Serving Pipeline

### Serving Architecture
**Serving Method**: [Real-time API / Batch prediction / Streaming]
**Infrastructure**: [SageMaker / Vertex AI / Custom / Serverless]
**Latency SLA**: [p50, p95, p99 targets]
**Throughput**: [Predictions per second]
**Fallback**: [Rule-based default / Previous model / Cached predictions]

### Feature Serving Flow
1. Request arrives with entity ID (user_id, product_id)
2. Online feature store lookup for real-time features
3. Feature vector assembled and validated
4. Model inference
5. Post-processing (thresholding, business rules, filtering)
6. Response returned

### A/B Testing Integration
- [ ] Traffic splitting configured (% per model variant)
- [ ] Metrics collection per variant (clicks, conversions, revenue)
- [ ] Statistical significance calculator integrated
- [ ] Automatic rollback on metric degradation

6. Data Quality Monitoring

## Data Quality Monitoring

### Automated Checks
| Check | Frequency | Threshold | Alert Channel |
|-------|-----------|-----------|--------------|
| Ingestion volume anomaly | Real-time | ± 30% from baseline | PagerDuty |
| Schema validation failures | Real-time | > 0.1% failure rate | Slack |
| Feature drift (PSI) | Daily | PSI > 0.2 | Email |
| Label distribution shift | Daily | KL divergence > 0.1 | Slack |
| Pipeline latency | Real-time | > 2x baseline | PagerDuty |
| Data freshness | Hourly | Exceeds SLA by 50% | PagerDuty |
| Null rate increase | Daily | > 2x baseline | Slack |

### Data Quality Dashboard
- Ingestion volume and error rates (real-time)
- Feature freshness per feature tier (hourly)
- Feature distribution histograms (daily)
- Training data quality scores (per training run)
- Pipeline DAG status and latency (real-time)

### Incident Response
| Severity | Criteria | Response Time | Action |
|----------|---------|---------------|--------|
| P0 | Complete pipeline failure | 15 min | Page on-call, failover to cached data |
| P1 | Feature freshness SLA breach | 1 hour | Investigate, extend serving cache |
| P2 | Data quality threshold exceeded | 4 hours | Investigate, pause retraining if needed |
| P3 | Minor anomaly detected | Next business day | Review and document |

7. Data Governance and Compliance

## Data Governance

### Privacy and Compliance
- [ ] PII fields identified and tagged in schema registry
- [ ] PII encryption at rest and in transit
- [ ] Data retention policies enforced per data type
- [ ] Right to deletion (GDPR/CCPA) propagated through pipeline
- [ ] Data access controls enforced per role
- [ ] Audit trail for all data access and transformations

### Data Lineage
**Lineage Tool**: [Apache Atlas / DataHub / Amundsen / Custom]
**Coverage**: [% of features with documented lineage]
**Update Frequency**: [Automatic / Per pipeline change]

### Access Control
| Role | Read Access | Write Access | Delete Access |
|------|-----------|-------------|--------------|
| Data Engineer | All pipeline data | Pipeline configs, schemas | With approval |
| ML Engineer | Features, training data | Model artifacts | None |
| Data Scientist | Features, training data, labels | Experiment configs | None |
| Product Manager | Dashboards, aggregated metrics | None | None |
| Auditor | Lineage, access logs, governance docs | None | None |

How to Use This Template

Start with freshness requirements. Interview your data science team: for each feature, what happens if the data is 1 hour stale? 1 day stale? 1 week stale? This drives your entire architecture. Real-time features need stream processing. Daily features can use batch pipelines.

Define quality gates before building. Agree on what "good enough" data quality means for each pipeline stage. Set thresholds, alerts, and runbooks before you have a production incident.

Build monitoring from day one. Do not treat monitoring as a post-launch addition. Data quality issues in AI systems are silent. The model still returns predictions, they are just wrong. See our guide to LLM evals for evaluation strategies.

Version everything. Schemas, features, transformations, and training datasets should all be versioned. When a model regresses, you need to trace back through every change in the pipeline.

Plan for backfills. You will need to reprocess historical data when you add new features, fix bugs in transformations, or change schemas. Design your pipeline to support backfills without disrupting production serving.

For a broader view of the AI product lifecycle, see the AI PM Handbook. The AI Build vs Buy assessment can help you decide which pipeline components to build versus buy from managed services.

AI Data Pipeline Template