Unemployment Trends in East Africa — Data Pipeline

A production-style data engineering pipeline that ingests live unemployment data from the World Bank API, validates structure and value bounds, transforms raw observations into an analytical dataset with year-over-year change calculations, and loads the result into a local DuckDB warehouse. The pipeline is orchestrated with Prefect, containerized with Docker for reproducible runs, and validated on every push via GitHub Actions.

Data source: World Bank Open Data API — Indicator SL.UEM.TOTL.ZS (unemployment, total % of labor force)
Geographic scope: East African Community (EAC) member states: Kenya, Uganda, Tanzania, Rwanda, Burundi, South Sudan, DR Congo

The Problem

The analysis workflow relied on a static CSV file manually exported from the World Bank at a single point in time. This approach introduced several engineering limitations:

Data becomes outdated: Analysis is frozen at the moment the CSV was downloaded; refreshing requires manual re-downloads
Manual data acquisition: No automated, repeatable process for acquiring fresh data
Weak reproducibility: Different analysts may use different CSV snapshots, leading to inconsistent results
Lack of automated validation: No structural checks on the raw data before transformation or analysis
Difficulty refreshing analysis: Updating the analysis requires repeating the entire manual workflow
External data reliability: No handling for API changes, pagination, or missing observations across countries and years
Missing observations: Not every country reports unemployment data for every year — the pipeline must distinguish between structurally invalid data and expected gaps in reporting

Moving to live API ingestion improves the pipeline by:

Current data: Each run retrieves the latest available observations from the World Bank
Automated acquisition: No manual downloads required
Reproducibility: The same code produces consistent results from the same API state
Explicit validation: Structural and value checks before transformation
Clear missing data strategy: Null unemployment values are allowed during validation where the source did not report data; rows without values are excluded during transformation

Important distinction: Live API ingestion does not guarantee complete reporting coverage. Source freshness and source completeness are separate concerns — the pipeline handles both explicitly.

Data Source and Scope

World Bank API

Endpoint: https://api.worldbank.org/v2/country/{country_codes}/indicator/SL.UEM.TOTL.ZS

Indicator: SL.UEM.TOTL.ZS — Unemployment, total (% of total labor force) (modeled ILO estimate)

API characteristics:

Public API — no API key required
Returns paginated JSON responses
Each observation includes: country code, country name, year, indicator value, unit of measurement
Not every country has observations for every year — missing data is expected and handled explicitly

Geographic Scope

The pipeline fetches data for all East African Community (EAC) member states:

Country	ISO Code
Kenya	KE
Uganda	UG
Tanzania	TZ
Rwanda	RW
Burundi	BI
South Sudan	SS
DR Congo	CD

Data Characteristics

Unit of measurement: Percentage of total labor force
Grain: One row per country per year (where data is reported)
Time coverage: All available years from the World Bank API (varies by country)
Missing observations: Some country-year combinations have no reported unemployment value — this is expected and handled explicitly

Architecture

┌─────────────────────┐
│  World Bank API     │
│  (external source)  │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│    Ingestion        │
│  (requests)         │
│  → raw JSON         │
│  → manifest         │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│    Validation       │
│  (pandera)          │
│  → schema gate      │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│  Transformation     │
│  (pandas)           │
│  → Parquet          │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│     Storage         │
│    (DuckDB)         │
│  → warehouse        │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│   Orchestration     │
│    (Prefect)        │
│  → DAG flow         │
└──────────┬──────────┘
           │
           ├──────────────────┐
           │                  │
           ▼                  ▼
┌─────────────────────┐  ┌─────────────────────┐
│    Containerization │  │       CI/CD         │
│      (Docker)       │  │  (GitHub Actions)   │
│  → reproducible run │  │  → pytest           │
└─────────────────────┘  └─────────────────────┘

Pipeline Stages

Stage	Responsibility	Technology	Output
Ingestion	Fetches unemployment data from World Bank API with pagination, writes raw JSON and manifest	`requests`	Raw JSON files + manifest JSON
Validation	Validates structure, country codes, year range, value bounds; allows nulls where countries didn't report	`pandera`	Validation gate (pass/fail)
Transformation	Drops unreported rows, computes year-over-year change per country, writes typed columnar output	`pandas`, `pyarrow`	Parquet files
Storage	Loads processed data into local analytical warehouse	`DuckDB`	DuckDB database with typed tables
Orchestration	Chains ingestion → validation → transformation → load into single flow with retries and hard validation gate	`Prefect`	Orchestrated DAG execution
Containerization	Packages full pipeline so it runs identically on any machine	`Docker`, `docker compose`	Reproducible container runtime
CI	Runs unit test suite and schema sanity check on every push/PR	`GitHub Actions`, `pytest`	Automated code validation

Pipeline Design

The pipeline is decomposed into independently runnable, independently tested modules. Each stage has a clear input, processing logic, validation, output, and failure behavior.

Ingestion

Input: Country codes (EAC member states), indicator code (SL.UEM.TOTL.ZS)

Processing:

Constructs World Bank API URL with country codes and indicator
Handles pagination automatically (World Bank API returns paginated responses)
Fetches all pages until complete dataset is retrieved
Parses JSON response and extracts observations
Writes raw JSON to data/raw/ with timestamped filename
Generates manifest JSON containing:
- Ingestion timestamp
- Indicator code
- Countries fetched
- Page count
- Total record count
- Source URL

Output: Raw JSON files + manifest JSON in data/raw/

Failure behavior: Pipeline halts if API request fails or pagination handling encounters an error. No downstream stages execute without valid raw data and manifest.

Why raw JSON is preserved: The raw layer provides traceability — the exact API response is captured and can be audited or reprocessed if needed.

Validation

Input: Raw JSON observations from ingestion stage

Processing:

Loads observations into a Pandas DataFrame
Applies a Pandera schema that enforces:
- Required columns: country_code, country_name, year, value
- Data types: All columns typed explicitly (string, integer, float)
- Country codes: Must match expected EAC member state codes (KE, UG, TZ, RW, BI, SS, CD)
- Country names: Must be non-empty strings
- Year range: Must be within reasonable bounds (e.g., 1960–2026)
- Value bounds: Unemployment percentage must be between 0 and 100 (inclusive) if present
- Nullability: value column is explicitly nullable — not every country reports every year
- Duplicate records: No duplicate (country_code, year) combinations

Output: Validation result (pass/fail)

Failure behavior: If validation fails, the pipeline halts immediately. No transformation or storage occurs.

Important distinction: A missing unemployment value from the World Bank is not automatically invalid source data. The schema allows nulls in the value column where the source did not report data. This distinguishes between:

Structurally invalid data: Wrong country code, year out of range, value outside 0–100 — fails validation
Missing reported observations: Valid record structure but no value reported — passes validation, handled during transformation

Transformation

Input: Validated DataFrame with raw observations

Processing:

Drop unreported rows: Removes rows where value is null (countries that didn't report for that year)
Sort: Orders by country code and year for consistent processing
Deduplication: Removes any duplicate (country_code, year) combinations (should not exist after validation, but defensive check)
Year-over-year calculation: For each country, computes the change in unemployment rate from the previous year:
- Formula: yoy_change = current_year_value - previous_year_value
- First year for each country has yoy_change = null (no prior year to compare)
Data type conversions: Ensures all columns are explicitly typed
Output column selection: Standardizes column order for downstream consistency

Output: Parquet file in data/processed/

Output fields:

Field	Transformation	Purpose
`country_code`	Direct from API	ISO country identifier
`country_name`	Direct from API	Human-readable country name
`year`	Direct from API	Observation year
`unemployment_rate`	Direct from API (`value`)	Unemployment % of labor force
`yoy_change`	`current_year_value - previous_year_value`	Year-over-year change in percentage points

Why missing observations are handled after validation: Structural validation ensures the data is well-formed before any filtering. Dropping unreported rows is a transformation decision, not a validation failure.

Storage

Input: Processed Parquet file

Processing:

Creates or recreates DuckDB database at data/warehouse/unemployment.duckdb
Loads Parquet into a table named unemployment_rates
Schema mirrors Parquet columns with explicit types
Table is fully replaced on each run (not incremental)

Output: DuckDB database with unemployment_rates table

Load method: Full reload — appropriate for this dataset size and single-node architecture.

Why DuckDB:

Local analytical workload: No server to manage; file-based database
Native Parquet support: Reads Parquet directly without ETL
Typed analytical storage: Enforces schema on load
Reproducibility: Database file is portable and versionable (though gitignored here)

Orchestration (Prefect)

Input: None (orchestrates all stages)

Flow structure:

Task 1: ingest_data() — fetches from World Bank API
Task 2: validate_data() — applies Pandera schema (hard gate)
Task 3: transform_data() — feature engineering and Parquet output
Task 4: load_to_warehouse() — DuckDB load

Dependencies: Sequential (ingest → validate → transform → load)

Retries: Configured with 2 retries for ingestion and storage tasks (transient network or file I/O errors)

Validation gate: If validate_data() fails, downstream tasks are skipped entirely

Failure behavior: Prefect marks the flow as failed, logs the error, and does not proceed to transformation or load.

Missing Data Strategy

This pipeline makes an explicit distinction between two types of "missing" data:

Structurally Invalid Data (Validation Failure)

Wrong country code (not in EAC scope)
Year outside reasonable range
Unemployment value outside 0–100 range
Duplicate (country, year) combinations
Missing required fields (country code, name, year)

Handling: Validation fails — pipeline halts, no downstream processing.

Missing Reported Observations (Expected Case)

Valid country code, year, and structure
Unemployment value is null (source did not report)

Handling:

Validation: Passes — null values are explicitly allowed in the schema
Transformation: Rows with null values are dropped before analytical storage
Rationale: These are not data quality failures — the World Bank simply doesn't have data for that country-year combination

Why this matters: Treating all nulls as validation failures would incorrectly flag expected gaps in reporting as errors. The pipeline distinguishes between "bad data" and "data not reported."

Dataset

Analytical Output

Source: World Bank Open Data API — Indicator SL.UEM.TOTL.ZS

Geographic scope: East African Community (EAC) member states (7 countries)

Time coverage: All available years from the World Bank API (varies by country)

Unit of measurement: Percentage of total labor force

Grain: One row per country per year (where data is reported)

Output format: Parquet file + DuckDB table

Row count: Varies by run — depends on how many country-year combinations have reported data

Data Characteristics

Unemployment rate: Modeled ILO estimate, total % of labor force
Year-over-year change: Absolute percentage-point change from prior year (null for first year per country)
Missing data: Some country-year combinations have no reported value — excluded from analytical output

Data Quality and Failure Handling

The pipeline enforces data quality at multiple stages:

Check	Stage	Enforcement
API response validation	Ingestion	HTTP status, pagination handling, JSON parsing
Schema validation	Validation	Pandera schema enforces column names, types, nullability
Country scope validation	Validation	Country codes must match EAC member states
Year validation	Validation	Years must be within reasonable bounds
Value bounds	Validation	Unemployment rate must be 0–100 if present
Nullable observations	Validation	Null values allowed where source didn't report
Duplicate detection	Validation	No duplicate (country_code, year) combinations
Transformation checks	Transformation	Drops unreported rows, defensive deduplication
Storage checks	Storage	Row count verified after DuckDB load
Pipeline failure	Orchestration	Failed tasks halt downstream execution; retries configured

Invalid data stops the pipeline: The validation gate ensures structurally invalid data never reaches transformation or storage.

Expected missing observations follow defined policy: Null values are allowed during validation but excluded during transformation — this is a deliberate design choice, not an oversight.

Project Structure

.
├── src/
│   ├── ingestion/          # World Bank API pull (paginated), raw JSON + manifest
│   ├── validation/         # Pandera schema definition and validation gate
│   ├── transformation/     # Cleaning, YoY feature, Parquet output
│   ├── storage/            # DuckDB load
│   └── pipeline/           # Prefect flow tying all stages together
├── tests/
│   ├── test_ingestion.py   # API response handling (mocked)
│   ├── test_validation.py  # Pandera schema tests
│   ├── test_transformation.py  # Feature engineering tests
│   ├── test_storage.py     # DuckDB load tests
│   └── test_pipeline.py    # End-to-end flow tests (synthetic data)
├── data/                   # Gitignored: raw/, processed/, warehouse/ (generated at runtime)
├── data-science/           # Original notebook, static CSV, and slide deck this project started from
├── Dockerfile              # Container image definition
├── docker-compose.yml      # Service orchestration
├── requirements.txt        # Python dependencies
└── .github/
    └── workflows/
        └── ci.yml          # GitHub Actions CI workflow

Notes on `data-science/`

The data-science/ directory contains the original analytical project from which this data engineering pipeline evolved:

Static CSV snapshot of World Bank data
Jupyter notebook with unemployment analysis
Slide deck presenting findings

This directory is preserved for reference but is not part of the live pipeline. The engineering project demonstrates the transformation from a static, manual workflow into a reproducible, automated pipeline.

Running the Pipeline

Prerequisites

Python 3.10+
Docker Desktop (optional, for containerized runs)
Internet access (required for World Bank API calls)

No API key required — the World Bank API is public.

1. Clone and configure

git clone <repository-url>
cd unemployment-ea-pipeline

2. Create virtual environment (recommended)

# Create virtual environment
python -m venv .venv

# Activate (Linux/macOS)
source .venv/bin/activate

# Activate (Windows)
.venv\Scripts\activate

3. Install dependencies

pip install -r requirements.txt

4. Run the pipeline locally

python -m src.pipeline.flow

The flow runs: ingestion → validation → transformation → load. If validation fails, the pipeline halts before transformation or storage.

5. Run the test suite

pytest tests/ -v

Unit tests use synthetic data and do not require live API access.

6. Run containerized

docker compose up --build

Docker Compose builds the image and runs the pipeline in a container with the same environment configuration.

Testing

The test suite is designed for speed, isolation, and reproducibility.

Test Module	Coverage
`test_ingestion.py`	API response handling with mocked responses
`test_validation.py`	Pandera schema validation (pass/fail cases)
`test_transformation.py`	Feature engineering, YoY calculations, missing value handling
`test_storage.py`	DuckDB load and row count verification
`test_pipeline.py`	End-to-end flow with synthetic data

Key properties:

Synthetic data: Tests do not depend on live API access or real data
Isolation: Each stage is tested independently
Speed: Full test suite runs in seconds
External dependency avoidance: CI can run tests without internet access or API availability

Orchestration (Prefect)

The Prefect flow orchestrates the pipeline rather than performing the transformations itself.

Flow Entry Point

src/pipeline/flow.py — defines the unemployment_data_pipeline() flow

Tasks

Task	Function	Purpose
Ingest	`ingest_data()`	Fetches data from World Bank API
Validate	`validate_data()`	Applies Pandera schema (hard gate)
Transform	`transform_data()`	Feature engineering and Parquet output
Load	`load_to_warehouse()`	DuckDB storage

Task Dependencies

Sequential: ingest_data → validate_data → transform_data → load_to_warehouse

Retries

Ingestion and storage tasks: 2 retries with exponential backoff (transient errors)
Validation and transformation: No retries (failures indicate data issues, not transient errors)

Failure Behavior

If validate_data() fails, downstream tasks are skipped
Prefect marks the flow as failed and logs the error
No partial data is written to storage

Containerization (Docker)

Dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY src/ src/
COPY data/ data/

CMD ["python", "-m", "src.pipeline.flow"]

docker-compose.yml

services:
  pipeline:
    build: .
    volumes:
      - ./data:/app/data

Why Containerization Matters

Reproducibility: Same Python version, dependencies, and environment on any machine
Isolation: No conflicts with host system packages
Portability: Runs identically on local machines, CI, or cloud infrastructure
Simplified onboarding: New engineers can run the pipeline with a single docker compose up command

Continuous Integration (GitHub Actions)

Workflow: `.github/workflows/ci.yml`

Triggers: Push to main, pull requests

Python version: 3.11

Steps:

Checkout repository
Install Python dependencies
Run pytest tests/ -v
Schema sanity check (validates Pandera schema loads without errors)

What CI does NOT do:

Does not run the live pipeline against the World Bank API
Does not fetch live data
Does not execute the full end-to-end flow

Rationale:

Repeatability: CI should not depend on external API availability
Test speed: Unit tests run in seconds; live API calls add latency
Reliability: CI should not fail due to temporary API outages
Independence: Code correctness is separate from data availability

Full pipeline runs against live data are a separate, deliberate concern requiring internet access and manual or scheduled execution.

Engineering Decisions

Live API Ingestion vs Static CSV

Decision: Fetch data from the World Bank API on each run rather than using a static CSV snapshot.

Reasons:

Current data: Each run retrieves the latest available observations
Repeatability: No manual downloads required
Reduced manual work: Automated acquisition eliminates human error
Source refreshes: Pipeline stays current as World Bank updates data

Trade-off: Requires internet access; API availability is an external dependency.

Important acknowledgment: Live ingestion does not guarantee complete reporting coverage — some country-year combinations will still have missing data.

Nullable Observations as a Deliberate Case

Decision: Allow null unemployment values during validation where the source did not report data.

Reasons:

Realistic data modeling: Not every country reports every year — this is expected
Distinguish failure modes: Structural invalidity vs. missing reporting are different problems
Explicit handling: Transformation drops unreported rows deliberately, not as a validation failure

Trade-off: Validation is more permissive, but transformation enforces the "must have value" rule before storage.

DuckDB for Local Analytical Storage

Decision: Use DuckDB rather than PostgreSQL, MySQL, or a cloud warehouse.

Reasons:

Local analytical workload: No server to manage; file-based database
Native Parquet support: Reads Parquet directly without ETL
Single-node architecture: Appropriate for this dataset size
Reproducibility: Database file is portable

Trade-off: Not suitable for multi-user concurrent writes or cloud-scale workloads.

Prefect for Orchestration

Decision: Use Prefect rather than Airflow or manual scripting.

Reasons:

Cross-platform execution: Runs on Windows, macOS, Linux without WSL or Docker
Task orchestration: Explicit DAG structure with dependencies
Retries: Built-in retry logic for transient errors
Failure visibility: Clear logging and task status
Lightweight setup: No database backend or web server required

Trade-off: Less mature ecosystem than Airflow for enterprise-scale deployments.

CI Without Live API Execution

Decision: GitHub Actions runs unit tests only, not the full pipeline against live data.

Reasons:

Repeatability: CI should not depend on external API availability
Test speed: Unit tests run in seconds
Reliability: CI should not fail due to temporary API outages
Independence: Code correctness is separate from data availability

Trade-off: CI does not validate end-to-end data flow — that requires manual or scheduled runs.

Reproducibility

To reproduce a pipeline run:

Python version: 3.10+ (tested on 3.11)
Dependencies: Install from requirements.txt
Internet access: Required for World Bank API calls
Docker (optional): docker compose up --build ensures identical environment

No API key required — the World Bank API is public.

Generated Artifacts

Artifact	Location	Git-tracked
Raw JSON	`data/raw/`	No
Manifest JSON	`data/raw/`	No
Processed Parquet	`data/processed/`	No
DuckDB database	`data/warehouse/`	No

Source Inputs

src/ — pipeline code
tests/ — test suite
requirements.txt — dependencies
Dockerfile, docker-compose.yml — containerization
.github/workflows/ci.yml — CI configuration
data-science/ — original analytical project (reference only)

Each run retrieves current available data from the World Bank API — results may vary slightly as the source is updated.

Data Flow

World Bank API
    ↓
Raw JSON (data/raw/)
    ↓
Pandera Validation
    ↓
Transformation (drop nulls, YoY calculation)
    ↓
Processed Parquet (data/processed/)
    ↓
DuckDB (data/warehouse/unemployment.duckdb)
    ↓
Analysis and downstream use

Downstream Use Cases

The resulting warehouse enables:

Unemployment trend analysis: Track unemployment rates over time by country
Country comparisons: Compare unemployment levels across EAC member states
Year-over-year analysis: Analyze changes in unemployment rates
Economic research: Study labor market dynamics in East Africa
Data science modelling: Export data for forecasting or classification models

The pipeline prepares the data; downstream consumers (analysts, researchers, models) use the warehouse as a trusted source.

Limitations and Future Improvements

Current Limitations

External API dependency: Pipeline requires World Bank API access and availability
No historical snapshots: Raw and processed files are overwritten on each run (no versioned snapshots)
No automated scheduling: Pipeline runs manually or via Docker Compose, not deployed to a production scheduler
Local DuckDB storage: Not suitable for multi-user concurrent access or cloud-scale workloads
No data observability platform: No monitoring, alerting, or drift detection beyond validation gate
No incremental loading: Full reload on each run (acceptable for this dataset size)
CI focused on unit tests: Does not validate end-to-end flow against live data

Potential Improvements

Historical versioning: Use DVC or similar to version raw and processed datasets
Scheduled orchestration: Deploy Prefect flow to Prefect Cloud or Airflow for automated periodic runs
Cloud warehouse: Migrate DuckDB to Snowflake, BigQuery, or Redshift for multi-user access
Data observability: Integrate with data quality monitoring tools (e.g., Great Expectations, Monte Carlo)
Incremental loading: Implement watermark-based incremental updates if historical data becomes important
API caching: Cache raw responses to reduce API calls and enable offline reprocessing

Technical Stack

Layer	Technology	Purpose
Language	Python 3.11	Pipeline implementation
Data ingestion	`requests`	Fetch data from World Bank API
Validation	`pandera`	Schema enforcement and data quality gate
Transformation	`pandas`, `pyarrow`	Feature engineering and Parquet output
Storage	`DuckDB`	Local analytical warehouse
Orchestration	`Prefect`	DAG-based workflow orchestration
Containerization	`Docker`, `docker compose`	Reproducible runtime environment
Testing	`pytest`	Unit and integration tests
CI	`GitHub Actions`	Automated code validation on push/PR

Acknowledgments

Data source: World Bank Open Data API — Indicator SL.UEM.TOTL.ZS (Unemployment, total % of total labor force) (modeled ILO estimate).

This project uses the World Bank data for educational and portfolio purposes. The pipeline implementation is original; the underlying data remains the property of the World Bank.

License

This project is licensed under the MIT License — see the LICENSE file for details.

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
data-science		data-science
src		src
tests		tests
.dockerignore		.dockerignore
.gitignore		.gitignore
Dockerfile		Dockerfile
LICENSE		LICENSE
README.md		README.md
docker-compose.yml		docker-compose.yml
pyproject.toml		pyproject.toml
requirements.txt		requirements.txt

Latest commit

History

Folders and files

Repository files navigation

Unemployment Trends in East Africa — Data Pipeline

The Problem

Data Source and Scope

World Bank API

Geographic Scope

Data Characteristics

Architecture

Pipeline Stages

Pipeline Design

Ingestion

Validation

Transformation

Storage

Orchestration (Prefect)

Missing Data Strategy

Structurally Invalid Data (Validation Failure)

Missing Reported Observations (Expected Case)

Dataset

Analytical Output

Data Characteristics

Data Quality and Failure Handling

Project Structure

Notes on data-science/

Running the Pipeline

Prerequisites

1. Clone and configure

2. Create virtual environment (recommended)

3. Install dependencies

4. Run the pipeline locally

5. Run the test suite

6. Run containerized

Testing

Orchestration (Prefect)

Flow Entry Point

Tasks

Task Dependencies

Retries

Failure Behavior

Containerization (Docker)

Dockerfile

docker-compose.yml

Why Containerization Matters

Continuous Integration (GitHub Actions)

Workflow: .github/workflows/ci.yml

Engineering Decisions

Live API Ingestion vs Static CSV

Nullable Observations as a Deliberate Case

DuckDB for Local Analytical Storage

Prefect for Orchestration

CI Without Live API Execution

Reproducibility

Generated Artifacts

Source Inputs

Data Flow

Downstream Use Cases

Limitations and Future Improvements

Current Limitations

Potential Improvements

Technical Stack

Acknowledgments

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages

Notes on `data-science/`

Workflow: `.github/workflows/ci.yml`