Intermediate

Data Pipeline

Build robust data pipelines for ML - covering ingestion, validation, feature stores, versioning, and orchestration.

Data Ingestion

Data ingestion is the first step in any ML pipeline. There are two primary patterns:

Batch Ingestion

Process data in large chunks on a schedule (hourly, daily, weekly). Suitable for most ML use cases where real-time data isn't required.

Python - Batch ingestion with Pandas

import pandas as pd
from sqlalchemy import create_engine

# Ingest from database
engine = create_engine("postgresql://user:pass@host/db")
df = pd.read_sql("SELECT * FROM transactions WHERE date >= '2025-01-01'", engine)

# Ingest from cloud storage
df_s3 = pd.read_parquet("s3://my-bucket/data/transactions.parquet")

# Ingest from API
import requests
response = requests.get("https://api.example.com/data", params={"limit": 10000})
df_api = pd.DataFrame(response.json()["results"])

Streaming Ingestion

Process data in real-time as it arrives. Required for use cases like fraud detection, recommendations, or live monitoring.

Python - Streaming with Apache Kafka

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'ml-features',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    features = message.value
    prediction = model.predict(features)
    publish_prediction(prediction)

Data Validation

Never trust incoming data blindly. Validate it before it enters your ML pipeline.

Great Expectations

Python - Data validation with Great Expectations

import great_expectations as gx

context = gx.get_context()

# Define expectations
validator = context.sources.pandas_default.read_dataframe(df)
validator.expect_column_values_to_not_be_null("user_id")
validator.expect_column_values_to_be_between("age", min_value=0, max_value=120)
validator.expect_column_values_to_be_in_set("status", ["active", "inactive", "pending"])
validator.expect_column_mean_to_be_between("purchase_amount", min_value=10, max_value=500)

# Run validation
results = validator.validate()
if not results.success:
    raise ValueError(f"Data validation failed: {results}")

TensorFlow Data Validation (TFDV)

Python - Schema validation with TFDV

import tensorflow_data_validation as tfdv

# Generate statistics from training data
train_stats = tfdv.generate_statistics_from_dataframe(train_df)

# Infer a schema from training data
schema = tfdv.infer_schema(train_stats)

# Validate new data against the schema
new_stats = tfdv.generate_statistics_from_dataframe(new_df)
anomalies = tfdv.validate_statistics(new_stats, schema)

# Check for anomalies
if anomalies.anomaly_info:
    print("Data anomalies detected!")
    tfdv.display_anomalies(anomalies)

Feature Stores

A feature store is a centralized repository for storing, managing, and serving ML features. It ensures consistency between training and serving.

Feature Store	Type	Best For
Feast	Open-source	Teams wanting full control, cloud-agnostic
Tecton	Managed	Enterprise teams needing real-time features
Hopsworks	Open-source / Managed	Full ML platform with feature store built-in
SageMaker Feature Store	Managed (AWS)	AWS-native ML workflows

Python - Feast feature store example

from feast import FeatureStore

store = FeatureStore(repo_path="feature_repo/")

# Define feature references
features = [
    "user_features:total_purchases",
    "user_features:avg_order_value",
    "user_features:days_since_last_purchase",
]

# Get features for training (historical)
training_df = store.get_historical_features(
    entity_df=entity_df,  # DataFrame with entity keys + timestamps
    features=features
).to_df()

# Get features for serving (online / real-time)
online_features = store.get_online_features(
    features=features,
    entity_rows=[{"user_id": 12345}]
).to_dict()

Data Versioning

Just like code, data should be versioned to ensure reproducibility.

Shell - Data versioning with DVC

# Initialize DVC in your Git repo
dvc init

# Track a data file
dvc add data/training_data.csv

# Push data to remote storage
dvc remote add -d myremote s3://my-bucket/dvc-storage
dvc push

# Switch to a different data version
git checkout v1.0
dvc checkout

# Compare data versions
dvc diff HEAD~1

ETL vs ELT

Aspect	ETL (Extract-Transform-Load)	ELT (Extract-Load-Transform)
Transform location	Before loading (staging area)	After loading (in data warehouse)
Best for	Structured data, known schemas	Large-scale, diverse data sources
Performance	Limited by staging compute	Leverages warehouse compute power
Flexibility	Schema must be defined upfront	Raw data preserved, transform as needed
Tools	Informatica, Talend, SSIS	dbt, Snowflake, BigQuery, Databricks

✅

For ML pipelines: ELT is generally preferred because it preserves raw data, allowing data scientists to experiment with different transformations without re-ingesting data.

Data Quality Monitoring

Continuously monitor data quality in production:

Completeness: Are all expected fields present? What's the null rate?
Freshness: Is data arriving on schedule? Are there delays?
Distribution: Have feature distributions shifted from training data?
Volume: Is the data volume within expected bounds?
Schema: Have column types or names changed unexpectedly?

Orchestration with Apache Airflow

Python - Airflow DAG for ML data pipeline

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'ml-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'ml_data_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=datetime(2025, 1, 1),
    catchup=False,
) as dag:

    ingest = PythonOperator(
        task_id='ingest_data',
        python_callable=ingest_from_sources,
    )

    validate = PythonOperator(
        task_id='validate_data',
        python_callable=run_data_validation,
    )

    transform = PythonOperator(
        task_id='transform_features',
        python_callable=compute_features,
    )

    store = PythonOperator(
        task_id='update_feature_store',
        python_callable=update_feast_store,
    )

    ingest >> validate >> transform >> store

💡

Alternatives to Airflow: Consider Prefect for a more Pythonic API, Dagster for data-aware orchestration with strong typing, or Mage for a modern, notebook-friendly pipeline builder.

← Previous ML Lifecycle Next → Model Training

Ready to Go Deeper?

Live instructor-led courses from our partners. Affiliate disclosure.

🧠

AI & ML Courses - 30% Off

Live instructor-led AI, machine learning, data science, and cloud courses for working professionals. Use code Limited30 at checkout.

Edureka

📊

DataCamp - AI & Data Science

Hands-on Python, machine learning, and AI courses with interactive exercises and real projects.

DataCamp

🎓

edX - Top AI Courses

University-level AI courses from MIT, Harvard, Stanford. Earn certificates that employers recognize.

edX