Building ETL Pipelines

Learn to build production-ready data pipelines with PyCharter's ETL system.

What You'll Learn

• Create pipelines from typed steps (dicts or YAML)
• Use built-in extractors (HTTP, file, database, cloud)
• Apply transformations (rename, filter, add fields)
• Load data to various destinations
• Handle errors and retries
• Use config-driven pipelines

Prerequisites

pip install pycharter[pipeline]

Part 1: Your First Pipeline

Steps in, records through

A pipeline is a list of typed steps — extract, transform, load — wired by input:. The same dict shape also lives in pipeline.yaml files (Part 5), so everything you learn here transfers to config-driven runs:

import pycharter

result = pycharter.run(
    {
        "name": "users_pipeline",
        "steps": [
            {
                "id": "fetch_users",
                "type": "extract",
                "source": {
                    "type": "http",
                    "url": "https://jsonplaceholder.typicode.com/users",
                },
            },
            {
                "id": "shape_users",
                "type": "transform",
                "input": "fetch_users",
                "operations": [{"rename": {"username": "user_name"}}],
            },
            {
                "id": "save_users",
                "type": "load",
                "input": "shape_users",
                "target": {
                    "type": "file",
                    "path": "output/users.json",
                    "format": "json",
                    "mode": "overwrite",
                },
            },
        ],
    }
)
print(f"Loaded {result.rows_loaded} rows in {result.duration_seconds:.2f}s")

Understanding the Result

The PipelineResult contains execution metrics:

result = pycharter.run(pipeline_config)

print(f"Rows extracted: {result.rows_extracted}")
print(f"Rows transformed: {result.rows_transformed}")
print(f"Rows loaded: {result.rows_loaded}")
print(f"Rows failed: {result.rows_failed}")
print(f"Duration: {result.duration_seconds:.2f}s")
print(f"Success: {result.success}")

if result.errors:
    for error in result.errors:
        print(f"Error: {error}")

Part 2: Extractors

HTTP Extractor

Extract data from REST APIs:

from pycharter import HTTPExtractor

# Simple GET request
extractor = HTTPExtractor(
    url="https://api.example.com/users",
    method="GET",
    headers={"Authorization": "Bearer ${API_KEY}"},
)

# With pagination
extractor = HTTPExtractor(
    url="https://api.example.com/users",
    pagination={"type": "offset", "param": "offset", "limit": 100},
)

# POST with body
extractor = HTTPExtractor(
    url="https://api.example.com/search", method="POST", body={"query": "active users"}
)

File Extractor

Extract from local files:

from pycharter import FileExtractor

# JSON file
extractor = FileExtractor(path="data/users.json", file_format="json")

# CSV file
extractor = FileExtractor(path="data/users.csv", file_format="csv")

# Parquet file
extractor = FileExtractor(path="data/users.parquet", file_format="parquet")

# Glob pattern (multiple files)
extractor = FileExtractor(path="data/*.json", file_format="json")

Database Extractor

Extract from SQL databases:

from pycharter import DatabaseExtractor

extractor = DatabaseExtractor(
    connection_string="postgresql://user:pass@localhost/db",
    query="SELECT * FROM users WHERE active = true",
)

# With parameters
extractor = DatabaseExtractor(
    connection_string="postgresql://user:pass@localhost/db",
    query="SELECT * FROM users WHERE created_at > %(start_date)s",
    params={"start_date": "2024-01-01"},
)

Cloud Storage Extractor

Extract from S3, GCS, or Azure:

from pycharter import CloudStorageExtractor

# AWS S3
extractor = CloudStorageExtractor(
    provider="s3",
    bucket="my-bucket",
    path="data/users.json",
    credentials={
        "aws_access_key_id": "${AWS_KEY}",
        "aws_secret_access_key": "${AWS_SECRET}",
    },
)

# Google Cloud Storage
extractor = CloudStorageExtractor(
    provider="gcs", bucket="my-bucket", path="data/users.json"
)

# Azure Blob Storage
extractor = CloudStorageExtractor(
    provider="azure", container="my-container", path="data/users.json"
)

Part 3: Transformers

Chain multiple transformers to process your data:

Rename Fields

from pycharter import Rename

# Rename single field
transform = Rename({"old_name": "new_name"})

# Rename multiple fields
transform = Rename(
    {"userName": "user_name", "userEmail": "email", "createdAt": "created_at"}
)

Filter Records

from pycharter import Filter

# Lambda filter
transform = Filter(lambda r: r.get("active", False))

# Multiple conditions
transform = Filter(lambda r: r["age"] >= 18 and r["status"] == "active")

Add Fields

from pycharter import AddField

# Static value
transform = AddField("source", "api")

# Dynamic expression
transform = AddField("processed_at", "now()")  # Current timestamp
transform = AddField("full_name", "concat(first_name, ' ', last_name)")

# From other fields
transform = AddField("is_adult", "age >= 18")

Drop Fields

from pycharter import Drop

# Drop single field
transform = Drop(["password"])

# Drop multiple fields
transform = Drop(["password", "ssn", "internal_id"])

Select Fields

from pycharter import Select

# Keep only these fields
transform = Select(["id", "name", "email"])

Convert Types

from pycharter import Convert

# Convert field values with callables
transform = Convert(
    {
        "age": int,
        "price": float,
        "active": bool,
    }
)

Custom Function

from pycharter import CustomFunction


# Apply custom logic
def enrich_user(record):
    record["display_name"] = f"{record['first_name']} {record['last_name']}"
    record["email_domain"] = record["email"].split("@")[1]
    return record


transform = CustomFunction(enrich_user)

Chaining Transformers

Transformer objects compose with | into a TransformerChain you can apply to records directly (useful in tests and custom functions). In a pipeline, the same operations are the transform step's ordered operations: list:

from pycharter import AddField, Convert, Drop, Filter, Rename

chain = (
    Rename({"userName": "user_name"})
    | Filter(lambda r: r.get("active"))
    | AddField("processed_at", "now()")
    | Drop(["password", "ssn"])
    | Convert({"age": int})
)
records = chain.transform([{"userName": "ada", "active": True, "age": "36"}])

- id: shape_users
  type: transform
  input: fetch_users
  operations:
    - rename: {userName: user_name}
    - add: {processed_at: now()}
    - drop: [password, ssn]
    - convert: {age: int}

Part 4: Loaders

PostgreSQL Loader

from pycharter import PostgresLoader

loader = PostgresLoader(
    connection_string="postgresql://user:pass@localhost/db",
    table="users",
    schema="public",
    mode="upsert",  # insert, upsert, replace
    match_on=["id"],
)

File Loader

from pycharter import FileLoader

# JSON output
loader = FileLoader(path="output/users.json", file_format="json")

# CSV output
loader = FileLoader(path="output/users.csv", file_format="csv")

# JSONL (JSON Lines) - good for streaming
loader = FileLoader(path="output/users.jsonl", file_format="jsonl")

# Parquet (compressed, columnar)
loader = FileLoader(path="output/users.parquet", file_format="parquet")

Cloud Storage Loader

from pycharter import CloudStorageLoader

loader = CloudStorageLoader(
    provider="s3", bucket="my-bucket", path="output/users.json", file_format="json"
)

Part 5: Config-Driven Pipelines

Define pipelines in YAML for easier management. Shipped starters and step snippets (snippets/*) plus pycharter pipeline init presets are documented in Templates overview. For diagrams of coerce / validate / quality at each step, see Contract → ETL → quality.

Directory structure

pipelines/
└── users/
    └── pipeline.yaml    # canonical: api_version, kind, metadata, spec

Scaffold a starter:

pycharter pipeline init pipelines/users --template http-to-postgres

Example pipeline.yaml

pipeline.yaml

api_version: pycharter.io/v2
kind: Pipeline
metadata:
  name: users
  namespace: examples
  version: "1.0.0"
  title: Users pipeline
spec:
  variables:
    API_KEY: ${API_KEY:?API_KEY is required}
    DATABASE_URL: postgresql://localhost/analytics

  steps:
    - id: extract_users
      type: extract
      source:
        type: http
        url: https://api.example.com/users
        method: GET
        headers:
          Authorization: "Bearer ${API_KEY}"
        pagination:
          type: offset
          param: page
          limit: 100

    - id: transform_users
      type: transform
      input: extract_users
      operations:
        - type: rename
          mapping:
            userName: user_name
            userEmail: email
        - type: filter
          expression: "active == true"
        - type: add
          fields:
            processed_at: "now()"
            source: api
        - type: drop
          fields: [password, internal_id]

    - id: load_users
      type: load
      input: transform_users
      target:
        type: postgres
        connection_string: "${DATABASE_URL}"
        table: users
        db_schema: public
        mode: upsert
        match_on: [id]

Step-body fragments (HTTP sources, Postgres targets, transform ops) live under data/templates/pipeline/snippets/ for copy-paste into steps:.

Run config-driven pipeline

import asyncio
from pycharter import Pipeline

pipeline = Pipeline.from_dir("pipelines/users/")

pipeline = Pipeline.from_dir(
    "pipelines/users/",
    variables={
        "API_KEY": "your-api-key",
        "DATABASE_URL": "postgresql://...",
    },
)

result = pipeline.run()

Part 6: Error Handling

Error Modes

from pycharter.shared.errors import ErrorMode, ErrorContext

# Strict: stop on first error (default)
result = await pipeline.arun(error_context=ErrorContext(mode=ErrorMode.STRICT))

# Lenient: log errors and continue
result = await pipeline.arun(error_context=ErrorContext(mode=ErrorMode.LENIENT))

# Collect: gather all errors without stopping
result = await pipeline.arun(error_context=ErrorContext(mode=ErrorMode.COLLECT))

# Check collected errors
if result.errors:
    for error in result.errors:
        print(f"Error: {error}")

Try-Except Pattern

from pycharter.shared.errors import PyCharterError, ConfigError

try:
    pipeline = Pipeline.from_dir("pipelines/users/")
    result = pipeline.run()
except ConfigError as e:
    print(f"Configuration error: {e}")
except PyCharterError as e:
    print(f"Pipeline error: {e}")

Part 7: Advanced Patterns

Validation During Load

Add contract validation before loading:

from pycharter import Pipeline, Validator

validator = Validator.from_file("contracts/user.yaml")


def validate_and_filter(records):
    valid_records = []
    for record in records:
        result = validator.validate(record)
        if result.is_valid:
            valid_records.append(result.data.model_dump())
    return valid_records


pipeline = Pipeline(extractor) | CustomFunction(validate_and_filter) | loader

Step contract: validate and quality

Config-driven pipelines can attach a contract per step. Use validate: true for per-record rules and quality: true for aggregate data_quality.rules on the batch:

contract:
  ref: { type: store, name: order, version: "1.0.0" }
  validate: true
  quality: true
  coerce: true
  on_error: fail

on_error controls what happens to a record that fails validation: fail aborts the run, warn logs and keeps it, skip drops it silently, and quarantine routes it to a dead-letter queue. Every DAG step contract using quarantine must declare a dlq target. This applies to existing manifests and direct execution as well as newly authored pipelines; a missing destination is a configuration error, including during a dry run:

contract:
  ref: { type: store, name: order, version: "1.0.0" }
  validate: true
  on_error: quarantine
  dlq:
    target:
      type: file
      path: ./out/orders.dlq.jsonl

Each quarantined record is appended to the JSONL file as a diagnostic envelope (payload, reason, error_message, contract, validation_errors) — the same shape a streaming worker's DLQ produces.

To migrate an older manifest that used quarantine without a destination, add the dlq block above. If discarding invalid records is intentional, change the policy to on_error: skip. A successful quarantine count means the DLQ sink acknowledged the write; it does not establish a transaction with the normal destination. Extraction previews report would_quarantine and leave the DLQ untouched. See pipeline outcomes and quarantine policy.

Load-step quality_checks (row count, null rate) are a separate pipeline quality feature. See Contract vs pipeline quality.

Parallel Extraction

Extract from multiple sources:

import asyncio
from pycharter import Pipeline

# Create multiple pipelines
pipeline1 = Pipeline.from_dir("pipelines/source1/")
pipeline2 = Pipeline.from_dir("pipelines/source2/")


# Run in parallel
async def run_parallel():
    results = await asyncio.gather(pipeline1.run(), pipeline2.run())
    return results


results = asyncio.run(run_parallel())

Incremental Loading

Track progress for incremental loads:

from datetime import datetime, timedelta

# Get last run timestamp (from your tracking system)
last_run = datetime(2024, 1, 1)

pipeline = Pipeline.from_dir(
    "pipelines/incremental/",
    variables={
        "LAST_RUN": last_run.isoformat(),
        "CURRENT_RUN": datetime.now().isoformat(),
    },
)
)

Exercises

1. Basic Pipeline: Create a pipeline that extracts users from JSONPlaceholder API, filters for users with .com emails, and saves to a JSON file.

2. Transform Pipeline: Build a pipeline that reads a CSV file, converts all field names to snake_case, adds a timestamp, and writes to JSONL.

3. Database Pipeline: Create a pipeline that extracts from one PostgreSQL table, transforms the data, and loads to another table.

4. Config-Driven: Convert one of the above pipelines to use YAML configuration files.

Next Steps

• Data Contracts & Validation - Add validation to your pipelines
• Quality Monitoring - Monitor pipeline data quality
• API Reference: Extractors - Complete extractor documentation
• API Reference: Transformers - Complete transformer documentation