Designing Data Pipelines

Section 5: Designing Data Migration Solutions. Lesson 27.

A data pipeline is a structured workflow that moves data from one place to another. In migration, a pipeline takes data from a source system, processes it, validates it and loads it into the target, for example moving customer data from MySQL to PostgreSQL.

The pipeline stages

Data source → Extract → Transform → Validate → Load → Monitor
                                        │
                                        └──▶ Error records (logged, fixable, reloadable)

A pipeline differs from a single SQL query. A proper pipeline must also handle:

• Data volume: batching, chunking and parallelism
• Data quality failures: validation and error records
• Dependencies: parents before children, jobs in order (lesson 29)
• Logging and monitoring: counts, timings and alerts
• Security: credentials, encryption and masking of sensitive fields
• Scheduling: when and how often it runs

Example: if customer data must be migrated every night, the pipeline should automatically extract the new or changed records, clean them, load them into the target and produce a success or failure report, with nobody running scripts by hand.

Three building blocks

Batch and streaming

A typical migration uses both: a batch pipeline moves the history, then a streaming or CDC pipeline keeps the target in sync until cutover (lesson 24).

Cloud pipeline services

• Google Cloud Dataflow: a fully managed platform for both batch and streaming processing, built on Apache Beam. Google positions it for real-time decision-making, with AI and ML integration, multimodal processing for generative AI, templates and notebooks, and built-in diagnostics and monitoring.
• Google Cloud Data Fusion: a managed, cloud-native data integration service for building and managing pipelines visually, without code. It is often used to bring data from siloed on-premises systems into Google Cloud and into data lakes. It is based on the open-source CDAP.
• Equivalents elsewhere: AWS Glue and Step Functions, Azure Data Factory, and Apache Airflow for orchestration (lesson 25).

Error handling and monitoring

Two components make a pipeline trustworthy.

Error handling: if a record fails because of a missing email, a duplicate ID, invalid data or a wrong data type, the pipeline must not ignore it silently. It should:

• Log the failed record
• Record the reason it failed
• Let the team fix it and reload it

Monitoring: after each run, report how many records were extracted, transformed, passed validation, failed and were loaded, plus timings. Alert someone when failures exceed a threshold or the run doesn't finish.

Hands-on: a pipeline in Google Colab

The lecture builds a small pipeline step by step in a Google Colab notebook with pandas: CSV or old database, then extract, clean and transform, validate, load, and report. Reconstructed below. You can run it in a new Colab notebook (Python 3 runtime) as one cell or cell by cell.

import pandas as pd

# Source data (stands in for a CSV file or an old database table)
source = pd.DataFrame({
    "customer_id": [1, 2, 3, 4, 5],
    "full_name":   ["Ali Khan", "Sara Ahmed", "Bilal Noor", "Hina Malik", "Omar Farooq"],
    "email":       ["ali@mail.com", "sara@mail.com", "bilal@mail.com", None, "omar@mail.com"],
    "phone":       ["0300-1111111", "0301-2222222", "0302-3333333", "0303-4444444", None],
    "city":        ["Lahore", "Karachi", "Islamabad", "Lahore", None],
})
print(source)

# 1. Extract: read from the source and work on a copy, never the original
extracted = source.copy()

# 2. Transform: split full_name into first_name and last_name, tidy strings
transformed = extracted.copy()
parts = transformed["full_name"].str.strip().str.split(" ", n=1, expand=True)
transformed["first_name"] = parts[0]
transformed["last_name"] = parts[1]
transformed["email"] = transformed["email"].str.strip().str.lower()
transformed = transformed.drop(columns=["full_name"])

# 3. Validate: required fields must be present; failures are kept, with a reason
required = ["email", "phone", "city"]
invalid_mask = transformed[required].isna().any(axis=1)
valid = transformed[~invalid_mask]
errors = transformed[invalid_mask].copy()
errors["error_reason"] = errors[required].isna().apply(
    lambda row: "missing: " + ", ".join(c for c in required if row[c]), axis=1)

# 4. Load: clean rows to the target, failed rows to an error file for review
valid.to_csv("new_customers.csv", index=False)
errors.to_csv("migration_errors.csv", index=False)
print("Pipeline completed.")

# 5. Monitor: a run summary per stage
summary = pd.DataFrame([
    {"stage": "extract",   "description": "records read from source",      "output": len(extracted)},
    {"stage": "transform", "description": "names split, strings cleaned",  "output": len(transformed)},
    {"stage": "validate",  "description": "passed / failed required checks", "output": f"{len(valid)} / {len(errors)}"},
    {"stage": "load",      "description": "rows written to new_customers.csv", "output": len(valid)},
])
print(summary.to_string(index=False))
print(errors[["customer_id", "error_reason"]])

Result: 5 records extracted, 5 transformed, 3 valid and 2 failed validation (one missing an email, one missing a phone and city). The clean rows go to new_customers.csv, and the failed rows go to migration_errors.csv with their reasons, ready to fix and reload. The lecture's sample data used different missing fields; the outcome is the same.

The steps are separate to make the architecture clear. In production each step would be a function or a task in an orchestrator such as Airflow, with the load step writing to PostgreSQL rather than a CSV:

from sqlalchemy import create_engine
target = create_engine("postgresql+psycopg://etl_user:***@pg-new:5432/crm")
valid.to_sql("customers", target, if_exists="append", index=False)
errors.to_sql("migration_errors", target, if_exists="append", index=False)

Pipeline design checklist

• Idempotent: re-running a batch doesn't duplicate rows (use upserts or keyed loads)
• Restartable: a failed run resumes from the last good checkpoint
• Observable: counts at every stage reconcile, and the run report is kept
• Safe: no credentials in code; sensitive fields masked outside production
• Ordered: dependencies respected (lesson 29)
• Tested: run against a sample, then a full dress rehearsal before cutover

Key takeaways

• A pipeline is the full path: source, extract, transform, validate, load, monitor.
• It handles volume, quality failures, dependencies, logging, security and scheduling, which a single query does not.
• Batch runs on a schedule; streaming runs continuously. Migrations often use both.
• Never drop failures silently: log them with reasons, then fix and reload.

Next: 28. Choosing Migration Tools.