Difficulty

Intermediate

Reward

300 XP · 🔌 Integration Engineer

Prerequisites

Python · PySpark · REST APIs

Primary Stack

Databricks · Delta Lake

Target

Snowflake

Development Task

DEV-003

REST API Integration Pipeline

Build a resilient external REST API ingestion pipeline that extracts paginated customer activity events, manages rate limits and network transient errors, validates raw payloads, and publishes curated analytical datasets to Snowflake.

Project Brief

Understand the business requirement and engineering challenge of integrating external APIs.

Business Scenario

A modern retail organization relies on a third-party customer engagement platform that records real-time user actions, web interactions, in-app purchases, and support events. This partner system exposes an external REST API for data extraction.

The analytics and customer experience teams require these activity streams to calculate customer engagement scores, conversion rates, and multi-channel marketing effectiveness. However, because the API is hosted externally, the data engineering team cannot control its availability, payload consistency, or response times.

Development Objective

Design and develop a production-ready ingestion pipeline that retrieves activity events across paginated endpoints, persists raw API responses for auditability, parses and validates complex JSON structures, removes duplicate events, standardizes Silver activity records, and delivers aggregated activity metrics to Snowflake.

Expected Outcome

Resilient API Ingestion

Safely retrieve paginated API data with token-based authentication, backoff retries, and rate-limiting controls.

Raw Payload Traceability

Persist untruncated JSON responses in the Bronze layer with request metadata for audit and replay.

Schema Validation

Parse nested JSON objects and quarantine malformed records without halting the pipeline execution.

Event Deduplication

Identify and eliminate duplicate activity records across overlapping extraction windows.

Silver Activity Dataset

Publish a clean, query-optimized Delta table containing standardized customer activity events.

Snowflake Target Delivery

Publish daily and user-level activity metrics to Snowflake for executive and BI consumption.

Developer Focus

The core challenge is balancing distributed processing with external API limits. You must structure ingestion to handle pagination, token auth, transient HTTP errors (5xx/429), and schema drift without hard-coding credentials or causing pipeline failure on dirty payloads.

Source System

Inspect the external API contract, pagination mechanics, and payload structure.

Source Overview

The source system is an external REST API endpoint providing paginated JSON responses of customer activity events. For development and testing, the API can be simulated or mocked using local mock servers or static JSON generators.

Source Type

External REST API (JSON)

Authentication

Bearer Token / API Key

Extraction Mode

Incremental Batch (Scheduled)

Pagination Style

Page + Limit / Cursor-based

Rate Limit

100 requests / min window

Domain

Customer Engagement & Activity

Example API Payload

Each API response contains a JSON payload with paging metadata and an array of activity records:

{
  "status": "success",
  "page": 1,
  "page_size": 100,
  "has_more": true,
  "data": [
    {
      "activity_id": "ACT-884920",
      "customer_id": "C1001",
      "activity_type": "purchase",
      "activity_timestamp": "2026-08-30T10:15:00Z",
      "amount": 125.50,
      "currency": "USD",
      "device_type": "mobile",
      "location": "Bengaluru",
      "metadata": {
        "channel": "app",
        "session_id": "S98762",
        "ip_address": "103.21.124.5"
      }
    }
  ]
}

Activity Schema Attributes

activity_id*

STRING

Unique business identifier for the activity event.

customer_id*

STRING

Unique identifier of the customer performing the activity.

activity_type*

STRING

Type of action (e.g., login, page_view, add_to_cart, purchase, review).

activity_timestamp*

TIMESTAMP

ISO 8601 UTC timestamp when the activity occurred.

amount

DECIMAL

Monetary transaction value if applicable; null for non-monetary actions.

currency

STRING

ISO currency code (e.g., USD, EUR, INR).

device_type

STRING

Originating client device type (mobile, desktop, tablet, smart-tv).

location

STRING

City or region associated with the activity session.

metadata

STRUCT/JSON

Nested properties including channel, session_id, and referral tags.

Known Source Constraints

Pagination & Empty Pages

The API may return empty pages or has_more=false to signal end of stream.

Rate Limiting (HTTP 429)

Exceeding request quotas yields 429 Too Many Requests with Retry-After header.

Transient 5xx Errors

Gateway timeouts (504) or Internal Errors (500) may occur intermittently.

Duplicate Activities Across Calls

Overlapping extraction intervals will yield duplicate activity_ids across pages.

Malformed JSON Payloads

Occasional corrupted responses or missing required fields must not crash ingestion.

Engineering Note

The source contract defines how the API behaves under both optimal and degraded conditions. Your pipeline must be resilient against network drops, authentication expiry, rate limits, and corrupted payloads.

Expected Architecture

Design a resilient multi-stage architecture from API ingestion to Snowflake delivery.

Target Architecture Flow

The pipeline decouples network extraction from distributed data processing. Raw API responses are landed immediately into Bronze storage, parsed and validated into Silver Delta tables, and aggregated into Gold analytics datasets.

External Partner API

REST / HTTPS Endpoints

Originating customer activity endpoint providing paginated JSON responses.

↓

API Ingestion Layer

Python / Requests / Token Auth

Manage authenticated requests, pagination loops, rate-limit backoff, and retry policies.

↓

Bronze — Raw JSON

Cloud Storage / Delta Lake

Persist untruncated JSON responses with extraction metadata (HTTP status, batch_id, extraction_time).

↓

Silver — Validated Activity

Databricks / PySpark / Delta Lake

Parse JSON, flatten nested metadata, validate schemas, deduplicate on activity_id, and quarantine invalid events.

↓

Gold — Activity Analytics

Databricks / PySpark

Calculate daily customer activity summaries, engagement scores, and conversion funnel metrics.

↓

Analytics Target

Snowflake

Publish curated activity tables and aggregation views for enterprise reporting.

Layer Responsibilities

Bronze Layer· Preserve Raw Responses

Stores complete JSON response bodies as received from the API, enabling complete replayability without re-hitting external endpoints.

Silver Layer· Normalize & Standardize

Extracts individual events from JSON arrays, applies typed schemas, eliminates duplicate activity_ids, and isolates bad records.

Gold Layer· Serve Analytical Metrics

Aggregates customer engagement counts, daily activity summaries by channel, and total spend by device type.

Architectural Expectations

Config & Credential Isolation

API keys, endpoints, and tokens must be loaded via environment variables or secret scopes, never hard-coded.

Idempotent Extraction

Rerunning an extraction batch must overwrite or merge Bronze files cleanly without causing downstream duplication.

Exponential Backoff

Transient 5xx and 429 errors must trigger jittered exponential backoff rather than immediate failing or rapid hammering.

Dead-Letter Quarantine

Malformed payloads or missing required keys must be routed to a quarantine dataset with error diagnostics.

Architecture Principle

Never perform complex business logic directly in the extraction client. Extract to Bronze first, then use PySpark for distributed validation, deduplication, and transformation.

Development Requirements

Implement the functional components required to extract, validate, transform, and publish API data.

Developer Responsibilities

The implementation must fulfill the following 15 engineering requirements across extraction, data quality, transformation, and target publishing.

API Client & Request Handling

Implement a robust HTTP client for external API communication.

Required Checklist

✓Construct parametrized GET requests with headers, query parameters, and timeout configs.

✓Support dynamic date filtering (start_date, end_date) for incremental windowing.

✓Enforce request timeout limits to prevent hanging connections.

Expected Outcome

Reliable communication channel with external REST endpoints.

Pagination Mechanism

Traverse paginated endpoints until all records for the extraction window are retrieved.

Required Checklist

✓Implement page-number or cursor-based pagination loop.

✓Detect end-of-stream via has_more flag, empty data array, or status code.

✓Track total pages extracted and total record counts per run.

Expected Outcome

Complete data retrieval across multiple pages without dropped batches.

Authentication & Secrets Separation

Securely manage API credentials and tokens.

Required Checklist

✓Externalize API tokens, client IDs, and secret keys from code.

✓Support token injection via Authorization Bearer headers.

✓Handle token expiration errors with structured failure reporting.

Expected Outcome

Secure, non-hardcoded authentication architecture.

Raw Bronze Persistence

Persist unmodified raw JSON payloads to cloud storage / Bronze Delta.

Required Checklist

✓Write raw JSON responses with partitioning by extraction_date.

✓Include ingestion metadata: batch_id, extraction_timestamp, source_endpoint, http_status.

✓Preserve full payload fidelity for auditability and recovery.

Expected Outcome

Full traceability and replay capability from Bronze raw storage.

JSON Parsing & Unnesting

Parse nested JSON arrays into structured PySpark DataFrames.

Required Checklist

✓Explode nested event arrays into individual row-level records.

✓Flatten nested metadata attributes (channel, session_id, ip_address).

✓Apply explicit PySpark StructType schema definitions.

Expected Outcome

Tabularized DataFrame with strongly-typed columns.

Schema Validation & Typing

Enforce strict data type casting and required field validation.

Required Checklist

✓Cast activity_timestamp to TIMESTAMP and amount to DECIMAL(12,2).

✓Validate mandatory presence of activity_id, customer_id, and activity_type.

✓Ensure currency values conform to 3-character ISO standards.

Expected Outcome

Schema conformance preventing silent data corruption.

Activity Deduplication

Prevent duplicate activity events across overlapping extractions.

Required Checklist

✓Deduplicate records using activity_id as the primary business key.

✓Use windowing (row_number over partition by activity_id order by timestamp desc) to pick latest record.

✓Ensure Silver Delta table is updated using idempotent MERGE operations.

Expected Outcome

Exactly-once business representation of customer actions.

Retry Policy & Transient Error Handling

Handle transient network failures and 5xx server responses gracefully.

Required Checklist

✓Implement retry decorator/loop with configurable max retries (e.g., 3-5 attempts).

✓Apply exponential backoff with randomized jitter.

✓Distinguish retryable errors (500, 502, 503, 504) from fatal errors (400, 401, 403, 404).

Expected Outcome

Pipeline resilience against momentary external network glitches.

Rate Limit Management

Comply with API provider rate limits to prevent 429 throttling.

Required Checklist

✓Inspect HTTP 429 response headers (Retry-After) and pause accordingly.

✓Implement client-side throttling / sleep intervals between pagination requests.

✓Log rate-limiting occurrences for capacity monitoring.

Expected Outcome

Controlled request velocity complying with provider SLA.

Silver Activity Transformation

Produce the clean, curated Silver customer activity Delta table.

Required Checklist

✓Standardize activity_type casing (lowercase trimmed strings).

✓Standardize device_type categories (mobile, desktop, tablet, unknown).

✓Derive activity_date from activity_timestamp for efficient partitioning.

Expected Outcome

Query-optimized, validated Silver Delta table.

Dead-Letter Quarantine

Isolate malformed or invalid records without pipeline termination.

Required Checklist

✓Route rows failing schema or null checks to a quarantine / bad-records Delta table.

✓Attach failure_reason, raw_payload, and rejection_timestamp.

✓Provide metrics on rejected vs accepted record counts.

Expected Outcome

Zero silent data loss with full visibility into rejected records.

Gold Activity Aggregation

Generate curated metrics and business KPIs from Silver records.

Required Checklist

✓Calculate daily active users (DAU) by activity_type and channel.

✓Aggregate total revenue, purchase count, and average order value per customer.

✓Summarize device and location distributions for marketing analytics.

Expected Outcome

Business-ready Gold tables optimized for BI tools.

Snowflake Target Publishing

Publish curated activity and aggregation tables to Snowflake.

Required Checklist

✓Publish Silver activity records to Snowflake FACT_CUSTOMER_ACTIVITY.

✓Publish Gold aggregated metrics to Snowflake AGG_DAILY_ACTIVITY_SUMMARY.

✓Ensure atomic, idempotent publishing avoiding duplicate target rows.

Expected Outcome

Synchronized Snowflake data mart accessible by enterprise analysts.

Operational Logging & Metrics

Provide comprehensive telemetry for every pipeline execution.

Required Checklist

✓Log start time, end time, pages extracted, records fetched, and duration.

✓Record counts for: Bronze inserted, Silver merged, Quarantined, and Gold published.

✓Log structured JSON error details on fatal failures.

Expected Outcome

Full operational observability for on-call engineers.

Configuration Management

Externalize all pipeline parameters across environments.

Required Checklist

✓Maintain YAML/JSON configuration files for dev, staging, and prod environments.

✓Externalize endpoint URLs, batch sizes, retry counts, storage paths, and Snowflake schemas.

Expected Outcome

Seamless deployment across environments without code alterations.

Implementation Freedom

You may choose whether to extract via Python requests before dispatching to Spark, or use custom Spark UDFs/datasources. Document your architectural choice and justify its performance and maintainability.

Testing Requirements

Validate pipeline resilience against pagination, network errors, malformed payloads, and duplicates.

Required Test Scenarios

The developer must validate both successful ingestion paths and failure-recovery modes.

Successful Multi-Page Extraction

Scenario

Simulate API endpoint returning 5 sequential pages of 100 records each with has_more=true/false.

Expected Result

Pipeline iterates through all 5 pages, landing exactly 500 records in Bronze and Silver.

Empty Page / End-of-Stream Handling

Scenario

Simulate API returning page 1 with empty data array [] and has_more=false.

Expected Result

Pipeline completes gracefully with 0 records extracted, without throwing null-pointer or indexing exceptions.

HTTP 4xx Fatal Error Handling

Scenario

Simulate HTTP 401 Unauthorized or 404 Not Found response from API.

Expected Result

Pipeline immediately logs clear fatal authentication/path error, halts execution, and does not perform wasteful retries.

HTTP 5xx Transient Server Error

Scenario

Simulate HTTP 503 Service Unavailable on attempt 1 and 2, followed by 200 OK on attempt 3.

Expected Result

Pipeline retries with exponential backoff, recovers on attempt 3, and successfully completes ingestion.

HTTP 429 Rate-Limit Throttling

Scenario

Simulate HTTP 429 Too Many Requests with Retry-After: 2 header.

Expected Result

Pipeline detects 429, sleeps for 2 seconds, retries the request, and proceeds without terminating.

Malformed JSON Payload

Scenario

Inject corrupted JSON strings or invalid syntax into a response body.

Expected Result

Ingestion captures raw text to quarantine log without crashing, continuing processing of valid records.

Missing Required Field Validation

Scenario

Provide records missing activity_id, customer_id, or activity_type.

Expected Result

Invalid records are detected and quarantined; valid records are successfully promoted to Silver.

Duplicate Activity Event Filtering

Scenario

Provide 2 identical activity_id records in page 1 and page 2.

Expected Result

Silver Delta table contains exactly 1 unique record for that activity_id after MERGE.

Token Expiration & Refresh Simulation

Scenario

Simulate token expiry mid-pagination yielding 401 on page 3.

Expected Result

Pipeline captures failure context, cleans up partial batch, and provides actionable error telemetry.

Pipeline Re-run / Idempotency

Scenario

Execute the pipeline twice for the exact same extraction time window.

Expected Result

Target Silver and Gold tables maintain identical row counts and metrics without duplicate inflation.

Gold Aggregation Reconciliation

Scenario

Calculate sum(amount) and count(activity_id) in Gold and compare against trusted Silver records.

Expected Result

Metrics reconcile 100% with Silver ground truth across all activity types and channels.

Snowflake Target Validation

Scenario

Query Snowflake FACT_CUSTOMER_ACTIVITY and AGG_DAILY_ACTIVITY_SUMMARY tables.

Expected Result

Snowflake tables match Silver/Gold counts, schemas, nullability constraints, and data values.

Testing Principle

External integrations will fail in production. Demonstrating how your pipeline handles 429s, 503s, and dirty payloads is just as important as verifying normal multi-page extraction.

Acceptance Criteria

Verify that the implementation satisfies the production Definition of Done.

Definition of Done

The task is complete when all 12 criteria are satisfied and backed by code, test evidence, and query outputs.

All API Pages Retrieved Completely

Pagination loop retrieves 100% of available pages for the target window without missing batches.

Transient API Failures Handled Safely

5xx and network drops trigger exponential backoff retries without crashing or infinitely looping.

Rate Limiting Complied With

HTTP 429 responses and client-side throttle rules prevent API ban or excessive failure rates.

Raw JSON Persisted & Traceable

Bronze layer contains unmodified raw JSON payloads with batch_id and extraction timestamps.

Schema Validation Enforced

Payloads are parsed into strongly-typed columns with type casting and null checks.

Dead-Letter Quarantine Active

Malformed payloads and records missing critical keys are segregated into a quarantine dataset.

Duplicate Activities Eliminated

Silver Delta table maintains exactly one record per activity_id using idempotent MERGE.

Silver Table Validated & Standardized

Activity types, device categories, and timestamps are cleaned and normalized.

Gold Aggregations Reconcile

Daily activity summaries and user engagement metrics match Silver source data perfectly.

Snowflake Target Synchronized

Curated tables are published to Snowflake and verify referential integrity and row counts.

No Hard-Coded Secrets

All API keys, endpoints, and credentials are configuration-driven and externalized.

Automated Tests Pass

Unit, integration, and mock API tests execute successfully with documented results.

Acceptance Rule

Code alone is insufficient. Deliverables must include test execution outputs, mock API scenarios, and query reconciliation results.

Developer Deliverables

Submit code, configuration, tests, and execution proof for review.

Required Deliverables

Provide the following 11 artifacts in your project repository:

API Ingestion Module

Python client implementation handling authentication, requests, pagination, and backoff retries.

Bronze Raw Persistence Logic

Code responsible for writing raw JSON payloads and extraction metadata to Bronze storage.

Silver Transformation & Deduplication

PySpark module parsing JSON, flattening structs, validating schemas, and merging into Silver Delta.

Quarantine / Bad-Records Handler

Module routing corrupted records to dead-letter storage with failure diagnostics.

Gold Aggregation Pipeline

PySpark logic creating daily active user and revenue aggregation tables.

Snowflake Publishing Logic

Connector or SQL staging scripts loading Silver and Gold tables into Snowflake.

Mock API / Simulator Script

Test harness or mock script simulating paginated, throttling, and failing API responses.

Automated Test Suite

Unit and integration tests covering multi-page, retry, rate-limit, and duplicate scenarios.

Execution & Telemetry Evidence

Logs and console outputs demonstrating multi-page extraction, retry recovery, and Snowflake sync.

Configuration Files

Externalized YAML/JSON configuration files for dev and prod environments.

Architecture Documentation

README documenting design decisions, pagination strategy, error handling, and recovery steps.

Submission Principle

Ensure your mock API harness allows another engineer to run your entire test suite without requiring real external credentials.

Engineering Constraints

Adhere to operational and technical boundaries for enterprise API integrations.

Required Boundaries

The solution must strictly comply with the following 10 constraints:

No Hard-Coded Credentials

API tokens, client secrets, and database credentials must be injected via secure environment variables or secret vaults.

No Infinite Retries

Retry loops must have hard maximum retry bounds (e.g., max 5 attempts) to prevent process deadlocks.

No Silent Data Loss

Records failing parsing or schema validation must be recorded in quarantine datasets rather than discarded.

Raw Payload Fidelity

Raw API JSON response bodies must be stored unmutated before any downstream schema transformations.

Idempotent Target Updates

Repeated executions of the same extraction batch must yield identical target tables without duplicates.

Rate-Limit Compliance

The extraction client must respect provider rate limits and sleep upon receiving HTTP 429 status.

Layer Responsibility Isolation

Network extraction, Bronze persistence, Silver cleansing, Gold analytics, and Snowflake loading must remain decoupled.

Driver Memory Protection

Avoid accumulating millions of raw API response objects in single-node driver memory; stream or batch write to storage.

Structured Operational Telemetry

Pipeline execution logs must emit structured operational metrics (records fetched, merged, failed, duration).

Production Code Quality

Code must be modularized into reusable classes/functions with proper type annotations and docstrings.

Constraint Notice

Violating any of these constraints (such as committing hard-coded tokens or dropping bad records silently) will result in immediate code review rejection.

Suggested Project Structure

Recommended repository layout for production REST API pipeline development.

Recommended Project Layout

Structure your repository to separate API client logic, PySpark transformation layers, tests, and environment configurations:

DEV-003-api-integration-pipeline/
│
├── README.md
│
├── config/
│   ├── dev.yaml
│   ├── staging.yaml
│   └── prod.yaml
│
├── src/
│   ├── api/
│   │   ├── client.py            # HTTP client, pagination & retry logic
│   │   ├── auth.py              # Token manager & auth headers
│   │   └── mock_server.py       # Local mock API for testing
│   │
│   ├── ingestion/
│   │   └── bronze_writer.py     # Raw JSON landing & metadata tracking
│   │
│   ├── transformations/
│   │   ├── silver_activity.py   # JSON parsing, unnesting & deduplication
│   │   ├── gold_analytics.py    # Daily summary & user KPI aggregation
│   │   └── quarantine.py        # Dead-letter quarantine handler
│   │
│   └── publishing/
│       └── snowflake_loader.py  # Snowflake staging & merge loader
│
├── notebooks/
│   ├── 01_api_extraction.py
│   ├── 02_silver_curation.py
│   └── 03_gold_publishing.py
│
├── tests/
│   ├── test_api_client.py
│   ├── test_pagination.py
│   ├── test_retries_backoff.py
│   ├── test_silver_parsing.py
│   └── test_deduplication.py
│
└── docs/
    ├── architecture.md
    └── api_contract.md

Module Responsibilities

src/api/

Manages HTTP communication, headers, pagination loops, rate-limiting, and mock simulation.

src/ingestion/

Persists raw API response JSON files to Bronze storage with extraction metadata.

src/transformations/

Contains PySpark schema parsing, unnesting, deduplication, and Gold aggregation logic.

src/publishing/

Manages Snowflake connections, staging tables, and idempotent MERGE updates.

tests/

Automated unit and integration test suite covering pagination, retries, schemas, and duplicates.

config/

Environment-specific endpoint URLs, batch sizes, retry parameters, and storage paths.

docs/

Architecture decisions, API contract definitions, and troubleshooting playbooks.

Design Rationale

Separating API extraction from Spark transformations ensures that changes to external API schemas or endpoint formats do not break downstream analytical aggregations.

Submission Checklist

Final quality checklist before submitting DEV-003 for peer engineering review.

Final Review Checklist

Verify every checklist item before submitting your development work:

✓

API Request & Pagination Implemented

HTTP client iterates through all available pages and captures end-of-stream correctly.

✓

Authentication & Config Externalized

No API tokens, client secrets, or private keys are hard-coded in source files.

✓

Bronze Raw JSON Persistence Verified

Raw API responses are stored unmutated with batch metadata in cloud storage.

✓

JSON Unnesting & Typing Complete

Nested event arrays and metadata structs are converted into strongly-typed columns.

✓

Deduplication Enforced

Overlapping extraction batches do not produce duplicate activity_id records in Silver.

✓

Retry & Backoff Verified

Transient 5xx errors and network drops trigger jittered exponential backoff.

✓

Rate-Limit (429) Handling Tested

Client detects 429 status and respects provider Retry-After backoff duration.

✓

Quarantine Dead-Letter Active

Malformed payloads are diverted to bad-records tables with rejection telemetry.

✓

Silver Table Validated

Silver Delta table contains clean, standardized, query-optimized activity records.

✓

Gold Aggregation Reconciles

Daily activity summaries and customer engagement metrics match Silver data 100%.

✓

Snowflake Sync Tested

Curated tables are loaded into Snowflake target tables without row inflation.

✓

Test Scenarios Executed

All 12 required test scenarios have been validated with logs/evidence recorded.

✓

Documentation Complete

README contains setup instructions, mock API usage, and architecture diagrams.

✓

Code Review Ready

Code is linted, formatted, modularized, and free of debug logs or temporary scripts.

Ready for Review

Submit DEV-003 only after the API client, Bronze raw storage, Silver Delta curation, Gold analytics, Snowflake sync, and test suites have been thoroughly validated.