ETL Pipeline

The saved excerpt is a snapshot from when this skill was listed. The external source remains the complete and most current version.

---
name: ETL Pipeline
description: Design and automate Extract, Transform, Load data pipelines for data integration and analytics
version: 1.0.0
author: Claude Office Skills
category: data
tags:
  - etl
  - data-pipeline
  - integration
  - analytics
  - automation
department: engineering
models:
  - claude-3-opus
  - claude-3-sonnet
  - gpt-4
mcp:
  server: data-mcp
  tools:
    - postgres_query
    - mysql_query
    - mongodb_aggregate
    - bigquery_load
    - snowflake_api
capabilities:
  - Data extraction from multiple sources
  - Data transformation and cleaning
  - Automated data loading
  - Pipeline orchestration
input:
  - Source database connections
  - Transformation rules
  - Target schema definitions
  - Schedule configurations
output:
  - Transformed datasets
  - Pipeline status reports
  - Data quality metrics
  - Error logs
languages:
  - en
related_skills:
  - sheets-automation
  - database-sync
  - api-integration
---

# ETL Pipeline

Comprehensive skill for designing and automating Extract, Transform, Load data pipelines.

## Pipeline Architecture

### Core ETL Flow

```
DATA PIPELINE ARCHITECTURE:
┌─────────────────────────────────────────────────────────┐
│                     EXTRACT                              │
├─────────┬─────────┬─────────┬─────────┬─────────────────┤
│ Postgres│  MySQL  │ MongoDB │  APIs   │  Files (CSV/JSON)│
└────┬────┴────┬────┴────┬────┴────┬────┴────────┬────────┘
     │         │         │         │              │
     └─────────┴─────────┴────┬────┴──────────────┘
                              ▼
┌─────────────────────────────────────────────────────────┐
│                    TRANSFORM                             │
│  • Clean & Validate    • Aggregate & Join               │
│  • Normalize           • Calculate Metrics              │
│  • Deduplicate         • Apply Business Rules           │
└────────────────────────┬────────────────────────────────┘
                         ▼
┌─────────────────────────────────────────────────────────┐
│                      LOAD                                │
├─────────────┬─────────────┬─────────────┬───────────────┤
│  BigQuery   │  Snowflake  │  Redshift   │  Data Lake    │
└─────────────┴─────────────┴─────────────┴───────────────┘
```

## Source Connectors

### Database Connections

```yaml
sources:
  postgres:
    type: postgresql
    host: db.example.com
    port: 5432
    database: production
    ssl: true
    extraction:
      method: incremental
      key: updated_at
      batch_size: 10000

  mysql:
    type: mysql
    host: mysql.example.com
    port: 3306
    database: analytics
    extraction:
      method: cdc
      binlog: true

  mongodb:
    type: mongodb
    connection_string: mongodb+srv://...
    database: app_data
    extraction:
      method: change_streams
```

### API Sources

```yaml
api_sources:
  stripe:
    type: rest_api
    base_url: https://api.stripe.com/v1
    auth: bearer_token
    endpoints:
      - /charges
      - /customers
      - /subscriptions
    pagination: cursor
    rate_limit: 100/second

  salesforce:
    type: salesforce
    instance_url: https://company.salesforce.com
    auth: oauth2
    objects:
      - Account
      - Opportunity
      - Contact
    bulk_api: true
```

## Transformation Layer

### Common Transformations

```python
# Data Cleaning
transformations = {
    "clean_nulls": {
        "operation": "fill_null",
        "columns": ["email", "phone"],
        "value": "unknown"
    },
    
    "standardize_dates": {
        "operation": "date_parse",
        "columns": ["created_at", "updated_at"],
        "format": "ISO8601"
    },
    
    "normalize_currency": {
        "operation": "convert_currency",
        "source_column": "amount",
        "currency_column": "currency",
        "target": "USD"
    },
    
    "deduplicate": {
        "operation": "distinct",
        "key_columns": ["customer_id", "transaction_id"],
        "keep": "latest"
    }
}
```

### Aggregation Rule