Ingesting Data

Data Ingestion Patterns

This skill provides patterns for getting data INTO systems from external sources.

When to Use This Skill

• Importing CSV, JSON, Parquet, or Excel files
• Loading data from S3, GCS, or Azure Blob storage
• Consuming REST/GraphQL API feeds
• Building ETL/ELT pipelines
• Database migration and CDC (Change Data Capture)
• Streaming data ingestion from Kafka/Kinesis

Ingestion Pattern Decision Tree

What is your data source?
├── Cloud Storage (S3, GCS, Azure) → See cloud-storage.md
├── Files (CSV, JSON, Parquet) → See file-formats.md
├── REST/GraphQL APIs → See api-feeds.md
├── Streaming (Kafka, Kinesis) → See streaming-sources.md
├── Legacy Database → See database-migration.md
└── Need full ETL framework → See etl-tools.md

Quick Start by Language

Python (Recommended for ETL)

dlt (data load tool) - Modern Python ETL:

import dlt

# Define a source
@dlt.source
def github_source(repo: str):
    @dlt.resource(write_disposition="merge", primary_key="id")
    def issues():
        response = requests.get(f"https://api.github.com/repos/{repo}/issues")
        yield response.json()
    return issues

# Load to destination
pipeline = dlt.pipeline(
    pipeline_name="github_issues",
    destination="postgres",  # or duckdb, bigquery, snowflake
    dataset_name="github_data"
)

load_info = pipeline.run(github_source("owner/repo"))
print(load_info)

Polars for file processing (faster than pandas):

import polars as pl

# Read CSV with schema inference
df = pl.read_csv("data.csv")

# Read Parquet (columnar, efficient)
df = pl.read_parquet("s3://bucket/data.parquet")

# Read JSON lines
df = pl.read_ndjson("events.jsonl")

# Write to database
df.write_database(
    table_name="events",
    connection="postgresql://user:pass@localhost/db",
    if_table_exists="append"
)

TypeScript/Node.js

S3 ingestion:

import { S3Client, GetObjectCommand } from "@aws-sdk/client-s3";
import { parse } from "csv-parse/sync";

const s3 = new S3Client({ region: "us-east-1" });

async function ingestFromS3(bucket: string, key: string) {
  const response = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: key }));
  const body = await response.Body?.transformToString();

  // Parse CSV
  const records = parse(body, { columns: true, skip_empty_lines: true });

  // Insert to database
  await db.insert(eventsTable).values(records);
}

API feed polling:

import { Hono } from "hono";

// Webhook receiver for real-time ingestion
const app = new Hono();

app.post("/webhooks/stripe", async (c) => {
  const event = await c.req.json();

  // Validate webhook signature
  const signature = c.req.header("stripe-signature");
  // ... validation logic

  // Ingest event
  await db.insert(stripeEventsTable).values({
    eventId: event.id,
    type: event.type,
    data: event.data,
    receivedAt: new Date()
  });

  return c.json({ received: true });
});

Rust

High-performance file ingestion:

use polars::prelude::*;
use aws_sdk_s3::Client;

async fn ingest_parquet(client: &Client, bucket: &str, key: &str) -> Result<DataFrame> {
    // Download from S3
    let resp = client.get_object()
        .bucket(bucket)
        .key(key)
        .send()
        .await?;

    let bytes = resp.body.collect().await?.into_bytes();

    // Parse with Polars
    let df = ParquetReader::new(Cursor::new(bytes))
        .finish()?;

    Ok(df)
}

Go

Concurrent file processing:

package main

import (
    "context"
    "encoding/csv"
    "github.com/aws/aws-sdk-go-v2/service/s3"
)

func ingestCSV(ctx context.Context, client *s3.Client, bucket, key string) error {
    resp, err := client.GetObject(ctx, &s3.GetObjectInput{
        Bucket: &bucket,
        Key:    &key,
    })
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    reader := csv.NewReader(resp.Body)
    records, err := reader.ReadAll()
    if err != nil {
        return err
    }

    // Batch insert to database
    return batchInsert(ctx, records)
}

Ingestion Patterns

1. Batch Ingestion (Files/Storage)

For periodic bulk loads:

Source → Extract → Transform → Load → Validate
  ↓         ↓          ↓         ↓        ↓
 S3      Download   Clean/Map  Insert   Count check

Key considerations:

• Use chunked reading for large files (>100MB)
• Implement idempotency with checksums
• Track file processing state
• Handle partial failures

2. Streaming Ingestion (Real-time)

For continuous data flow:

Source → Buffer → Process → Load → Ack
  ↓        ↓         ↓        ↓      ↓
Kafka   In-memory  Transform  DB   Commit offset

Key considerations:

• At-least-once vs exactly-once semantics
• Backpressure handling
• Dead letter queues for failures
• Checkpoint management

3. API Polling (Feeds)

For external API data:

Schedule → Fetch → Dedupe → Load → Update cursor
   ↓         ↓        ↓       ↓         ↓
 Cron     API call  By ID   Insert   Last timestamp

Key considerations:

• Rate limiting and backoff
• Incremental loading (cursors, timestamps)
• API pagination handling
• Retry with exponential backoff

4. Change Data Capture (CDC)

For database replication:

Source DB → Capture changes → Transform → Target DB
    ↓             ↓               ↓            ↓
 Postgres    Debezium/WAL      Map schema   Insert/Update

Key considerations:

• Initial snapshot + streaming changes
• Schema evolution handling
• Ordering guarantees
• Conflict resolution

Library Recommendations

Reference Documentation

• references/cloud-storage.md - S3, GCS, Azure Blob patterns
• references/file-formats.md - CSV, JSON, Parquet, Excel handling
• references/api-feeds.md - REST polling, webhooks, GraphQL subscriptions
• references/streaming-sources.md - Kafka, Kinesis, Pub/Sub
• references/database-migration.md - Schema migration, CDC patterns
• references/etl-tools.md - dlt, Meltano, Airbyte, Fivetran

Scripts

• scripts/validate_csv_schema.py - Validate CSV against expected schema
• scripts/test_s3_connection.py - Test S3 bucket connectivity
• scripts/generate_dlt_pipeline.py - Generate dlt pipeline scaffold

Chaining with Database Skills

After ingestion, chain to appropriate database skill:

For vector databases, chain through ai-data-engineering for embedding:

ingesting-data → ai-data-engineering → databases-vector

API Feed Ingestion

Table of Contents

• REST API Polling
• Python with httpx (async)
• TypeScript with fetch
• Webhook Receivers
• Python (FastAPI)
• TypeScript (Hono)
• GraphQL Subscriptions
• Python (gql)
• Rate Limiting & Backoff
• Best Practices

REST API Polling

Python with httpx (async)

import httpx
import asyncio
from datetime import datetime, timedelta

async def poll_api(
    url: str,
    interval_seconds: int = 60,
    cursor_field: str = "updated_at"
):
    """Poll API with incremental loading."""
    async with httpx.AsyncClient() as client:
        cursor = load_cursor()  # Load from DB/file

        while True:
            params = {cursor_field + "_gt": cursor} if cursor else {}

            response = await client.get(url, params=params)
            response.raise_for_status()

            data = response.json()
            if data["items"]:
                await process_items(data["items"])
                cursor = max(item[cursor_field] for item in data["items"])
                save_cursor(cursor)

            await asyncio.sleep(interval_seconds)

# With pagination
async def fetch_all_pages(url: str, page_size: int = 100):
    """Fetch all pages from paginated API."""
    async with httpx.AsyncClient() as client:
        page = 1
        while True:
            response = await client.get(url, params={
                "page": page,
                "per_page": page_size
            })
            data = response.json()

            yield data["items"]

            if len(data["items"]) < page_size:
                break
            page += 1

TypeScript with fetch

async function* fetchWithPagination(baseUrl: string, pageSize = 100) {
  let cursor: string | undefined;

  while (true) {
    const url = new URL(baseUrl);
    url.searchParams.set("limit", String(pageSize));
    if (cursor) url.searchParams.set("cursor", cursor);

    const response = await fetch(url);
    const data = await response.json();

    yield data.items;

    cursor = data.next_cursor;
    if (!cursor) break;
  }
}

// Usage
for await (const batch of fetchWithPagination("https://api.example.com/items")) {
  await processBatch(batch);
}

Webhook Receivers

Python (FastAPI)

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib

app = FastAPI()

@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")

    # Verify signature
    if not verify_stripe_signature(payload, sig_header):
        raise HTTPException(status_code=400, detail="Invalid signature")

    event = await request.json()

    # Idempotency check
    if await is_processed(event["id"]):
        return {"status": "already_processed"}

    # Process and store
    await store_event(event)
    await mark_processed(event["id"])

    return {"status": "ok"}

def verify_stripe_signature(payload: bytes, sig_header: str) -> bool:
    secret = os.environ["STRIPE_WEBHOOK_SECRET"]
    timestamp, signature = parse_stripe_header(sig_header)

    signed_payload = f"{timestamp}.{payload.decode()}"
    expected = hmac.new(
        secret.encode(),
        signed_payload.encode(),
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(signature, expected)

TypeScript (Hono)

import { Hono } from "hono";
import { createHmac, timingSafeEqual } from "crypto";

const app = new Hono();

app.post("/webhooks/github", async (c) => {
  const payload = await c.req.text();
  const signature = c.req.header("x-hub-signature-256");

  // Verify
  const expected = `sha256=${createHmac("sha256", process.env.GITHUB_SECRET!)
    .update(payload)
    .digest("hex")}`;

  if (!timingSafeEqual(Buffer.from(signature!), Buffer.from(expected))) {
    return c.json({ error: "Invalid signature" }, 401);
  }

  const event = JSON.parse(payload);
  await db.insert(githubEvents).values({
    eventId: c.req.header("x-github-delivery"),
    eventType: c.req.header("x-github-event"),
    payload: event,
    receivedAt: new Date()
  });

  return c.json({ received: true });
});

GraphQL Subscriptions

Python (gql)

from gql import Client, gql
from gql.transport.websockets import WebsocketsTransport

transport = WebsocketsTransport(
    url="wss://api.example.com/graphql",
    headers={"Authorization": f"Bearer {token}"}
)

async with Client(transport=transport) as session:
    subscription = gql("""
        subscription {
            orderCreated {
                id
                customer { name }
                items { product quantity }
                total
            }
        }
    """)

    async for result in session.subscribe(subscription):
        await process_order(result["orderCreated"])

Rate Limiting & Backoff

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=60)
)
async def fetch_with_retry(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            await asyncio.sleep(retry_after)
            raise Exception("Rate limited")
        response.raise_for_status()
        return response.json()

Best Practices

1. Always use idempotency keys - Prevent duplicate processing 2. Verify webhook signatures - Security is critical 3. Implement cursor-based pagination - More reliable than offset 4. Store raw payloads - Debug and replay capability 5. Use exponential backoff - Be a good API citizen

Cloud Storage Ingestion Patterns

Table of Contents

• S3 (AWS)
• Python with boto3
• TypeScript with AWS SDK v3
• GCS (Google Cloud)
• Python with gcsfs
• Azure Blob Storage
• Python with adlfs
• Best Practices

S3 (AWS)

Python with boto3

import boto3
import polars as pl
from io import BytesIO

s3 = boto3.client('s3')

def list_and_ingest(bucket: str, prefix: str):
    """List objects and ingest all matching files."""
    paginator = s3.get_paginator('list_objects_v2')

    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get('Contents', []):
            key = obj['Key']
            if key.endswith('.parquet'):
                ingest_parquet(bucket, key)

def ingest_parquet(bucket: str, key: str) -> pl.DataFrame:
    """Ingest Parquet file from S3."""
    response = s3.get_object(Bucket=bucket, Key=key)
    return pl.read_parquet(BytesIO(response['Body'].read()))

# Stream large files
def stream_csv(bucket: str, key: str, chunk_size: int = 10000):
    """Stream CSV in chunks for memory efficiency."""
    response = s3.get_object(Bucket=bucket, Key=key)

    reader = pl.read_csv_batched(response['Body'], batch_size=chunk_size)
    while True:
        batch = reader.next_batches(1)
        if not batch:
            break
        yield batch[0]

TypeScript with AWS SDK v3

import { S3Client, GetObjectCommand, ListObjectsV2Command } from "@aws-sdk/client-s3";
import { Readable } from "stream";

const s3 = new S3Client({ region: process.env.AWS_REGION });

async function* listObjects(bucket: string, prefix: string) {
  let continuationToken: string | undefined;

  do {
    const response = await s3.send(new ListObjectsV2Command({
      Bucket: bucket,
      Prefix: prefix,
      ContinuationToken: continuationToken
    }));

    for (const obj of response.Contents ?? []) {
      yield obj;
    }

    continuationToken = response.NextContinuationToken;
  } while (continuationToken);
}

async function downloadAsStream(bucket: string, key: string): Promise<Readable> {
  const response = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: key }));
  return response.Body as Readable;
}

GCS (Google Cloud)

Python with gcsfs

import gcsfs
import polars as pl

fs = gcsfs.GCSFileSystem(project='my-project')

# Direct read with Polars
df = pl.read_parquet('gs://bucket/path/data.parquet')

# List and process
files = fs.glob('gs://bucket/data/*.csv')
for file in files:
    with fs.open(file) as f:
        df = pl.read_csv(f)
        process(df)

Azure Blob Storage

Python with adlfs

import adlfs
import polars as pl

fs = adlfs.AzureBlobFileSystem(
    account_name='storageaccount',
    account_key='...'  # or use managed identity
)

# Read directly
df = pl.read_parquet('abfs://container/path/data.parquet')

Best Practices

1. Use appropriate file formats:

• Parquet for analytics (columnar, compressed)
• JSON Lines for streaming/logs
• CSV only for interchange with external systems

2. Implement resumable downloads:

   def download_with_resume(bucket, key, local_path):
       existing_size = os.path.getsize(local_path) if os.path.exists(local_path) else 0
       response = s3.get_object(
           Bucket=bucket,
           Key=key,
           Range=f'bytes={existing_size}-'
       )
       with open(local_path, 'ab') as f:
           f.write(response['Body'].read())

3. Track processed files:

   def mark_processed(key: str):
       db.execute(
           "INSERT INTO processed_files (key, processed_at) VALUES (?, ?)",
           (key, datetime.utcnow())
       )

4. Handle rate limits:

• Use exponential backoff
• Implement request throttling
• Consider S3 Inventory for large bucket listings

Database Migration & CDC

Table of Contents

• Schema Migration Tools
• Python (Alembic)
• TypeScript (Drizzle)
• Bulk Data Migration
• Python - Table to Table
• With Transformation
• Change Data Capture (CDC)
• PostgreSQL Logical Replication
• Debezium (Docker)
• Initial Load + CDC Pattern
• Data Validation

Schema Migration Tools

Python (Alembic)

# alembic/versions/001_initial.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('email', sa.String(255), nullable=False, unique=True),
        sa.Column('created_at', sa.DateTime, server_default=sa.func.now())
    )
    op.create_index('idx_users_email', 'users', ['email'])

def downgrade():
    op.drop_table('users')

TypeScript (Drizzle)

// drizzle/0001_initial.ts
import { pgTable, serial, varchar, timestamp } from "drizzle-orm/pg-core";

export const users = pgTable("users", {
  id: serial("id").primaryKey(),
  email: varchar("email", { length: 255 }).notNull().unique(),
  createdAt: timestamp("created_at").defaultNow()
});

// Run migration
import { migrate } from "drizzle-orm/node-postgres/migrator";
await migrate(db, { migrationsFolder: "./drizzle" });

Bulk Data Migration

Python - Table to Table

import polars as pl
from sqlalchemy import create_engine

source = create_engine("postgresql://source_db")
target = create_engine("postgresql://target_db")

def migrate_table(table_name: str, batch_size: int = 10000):
    """Migrate table in batches."""
    offset = 0

    while True:
        # Read batch from source
        query = f"SELECT * FROM {table_name} ORDER BY id LIMIT {batch_size} OFFSET {offset}"
        df = pl.read_database(query, source)

        if len(df) == 0:
            break

        # Write to target
        df.write_database(
            table_name=table_name,
            connection=target,
            if_table_exists="append"
        )

        offset += batch_size
        print(f"Migrated {offset} rows...")

With Transformation

def migrate_with_transform(source_table: str, target_table: str):
    """Migrate with schema transformation."""
    df = pl.read_database(f"SELECT * FROM {source_table}", source)

    # Transform
    df = df.rename({"old_column": "new_column"})
    df = df.with_columns([
        pl.col("amount").cast(pl.Decimal(10, 2)),
        pl.col("created_at").str.to_datetime()
    ])
    df = df.drop("deprecated_field")

    # Write
    df.write_database(target_table, target, if_table_exists="replace")

Change Data Capture (CDC)

PostgreSQL Logical Replication

import psycopg2
from psycopg2.extras import LogicalReplicationConnection

def consume_changes(slot_name: str = "my_slot"):
    conn = psycopg2.connect(
        connection_factory=LogicalReplicationConnection,
        dsn="postgresql://localhost/mydb"
    )
    cursor = conn.cursor()

    # Create replication slot
    cursor.create_replication_slot(slot_name, output_plugin='pgoutput')

    # Start replication
    cursor.start_replication(slot_name=slot_name)

    def consume(msg):
        # Parse logical replication message
        change = parse_pgoutput(msg.payload)
        process_change(change)
        msg.cursor.send_feedback(flush_lsn=msg.data_start)

    cursor.consume_stream(consume)

Debezium (Docker)

# docker-compose.yml
version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on: [zookeeper]
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092

  debezium:
    image: debezium/connect:2.4
    depends_on: [kafka]
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: debezium_config
      OFFSET_STORAGE_TOPIC: debezium_offsets
      STATUS_STORAGE_TOPIC: debezium_status

# Register Debezium connector
import requests

connector_config = {
    "name": "postgres-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname": "postgres",
        "database.port": "5432",
        "database.user": "debezium",
        "database.password": "secret",
        "database.dbname": "mydb",
        "topic.prefix": "mydb",
        "table.include.list": "public.users,public.orders",
        "plugin.name": "pgoutput"
    }
}

requests.post(
    "http://localhost:8083/connectors",
    json=connector_config
)

Initial Load + CDC Pattern

async def initial_load_with_cdc(table: str):
    """
    1. Start CDC capture
    2. Take snapshot
    3. Apply CDC changes after snapshot
    """
    # Mark CDC start position
    cdc_start = await get_current_lsn()

    # Full table snapshot
    snapshot_df = pl.read_database(f"SELECT * FROM {table}", source)
    await write_to_target(snapshot_df, table)

    # Apply buffered CDC changes
    changes = await get_changes_since(cdc_start)
    for change in changes:
        await apply_change(change)

    # Continue with live CDC
    await consume_cdc_stream()

Data Validation

def validate_migration(source_table: str, target_table: str) -> dict:
    """Validate migration completeness."""
    source_count = pl.read_database(
        f"SELECT COUNT(*) as cnt FROM {source_table}", source
    )["cnt"][0]

    target_count = pl.read_database(
        f"SELECT COUNT(*) as cnt FROM {target_table}", target
    )["cnt"][0]

    # Sample comparison
    source_sample = pl.read_database(
        f"SELECT * FROM {source_table} ORDER BY RANDOM() LIMIT 100", source
    )
    target_sample = pl.read_database(
        f"SELECT * FROM {target_table} WHERE id IN ({','.join(map(str, source_sample['id']))})", target
    )

    return {
        "source_count": source_count,
        "target_count": target_count,
        "count_match": source_count == target_count,
        "sample_match": source_sample.equals(target_sample)
    }

ETL Tools Reference

Table of Contents

• dlt (data load tool) - Recommended
• Installation
• Basic Pipeline
• Incremental Loading
• Transformations with dbt
• Meltano (ELT Platform)
• Installation
• Add Extractors/Loaders
• Configuration
• Run Pipeline
• Dagster (Orchestration + ELT)
• Assets-based Pipeline
• Airbyte (Low-Code ELT)
• Docker Setup
• API Configuration
• Tool Selection Guide
• Comparison Matrix

dlt (data load tool) - Recommended

Modern Python-first ETL with automatic schema evolution.

Installation

pip install dlt[postgres]  # or dlt[duckdb], dlt[bigquery], etc.

Basic Pipeline

import dlt

# Define source with resources
@dlt.source
def api_source(api_key: str):
    @dlt.resource(write_disposition="merge", primary_key="id")
    def users():
        response = requests.get(
            "https://api.example.com/users",
            headers={"Authorization": f"Bearer {api_key}"}
        )
        yield response.json()

    @dlt.resource(write_disposition="append")
    def events():
        response = requests.get("https://api.example.com/events")
        yield response.json()

    return users, events

# Create pipeline
pipeline = dlt.pipeline(
    pipeline_name="api_pipeline",
    destination="postgres",
    dataset_name="raw_data"
)

# Run
load_info = pipeline.run(api_source(os.environ["API_KEY"]))
print(load_info)

Incremental Loading

@dlt.source
def incremental_source():
    @dlt.resource(primary_key="id")
    def orders(
        updated_at=dlt.sources.incremental("updated_at", initial_value="2024-01-01")
    ):
        # Only fetch records after last updated_at
        response = requests.get(
            f"https://api.example.com/orders?updated_after={updated_at.last_value}"
        )
        yield response.json()

    return orders

Transformations with dbt

# Run dbt models after loading
pipeline.run(source)
pipeline.run(
    dbt_package="./dbt_project",
    select=["staging", "marts"]
)

Meltano (ELT Platform)

Singer-based with 500+ connectors.

Installation

pip install meltano
meltano init my_project
cd my_project

Add Extractors/Loaders

# Add extractors (sources)
meltano add extractor tap-github
meltano add extractor tap-salesforce

# Add loaders (destinations)
meltano add loader target-postgres
meltano add loader target-snowflake

Configuration

# meltano.yml
plugins:
  extractors:
    - name: tap-github
      config:
        repository: owner/repo
        start_date: '2024-01-01'

  loaders:
    - name: target-postgres
      config:
        host: localhost
        port: 5432
        database: warehouse

Run Pipeline

meltano run tap-github target-postgres

Dagster (Orchestration + ELT)

Assets-based Pipeline

from dagster import asset, Definitions
from dagster_duckdb import DuckDBResource

@asset
def raw_orders(duckdb: DuckDBResource):
    """Load orders from S3."""
    with duckdb.get_connection() as conn:
        conn.execute("""
            CREATE TABLE raw_orders AS
            SELECT * FROM read_parquet('s3://bucket/orders/*.parquet')
        """)

@asset(deps=[raw_orders])
def cleaned_orders(duckdb: DuckDBResource):
    """Clean and transform orders."""
    with duckdb.get_connection() as conn:
        conn.execute("""
            CREATE TABLE cleaned_orders AS
            SELECT
                id,
                customer_id,
                CAST(amount AS DECIMAL(10,2)) as amount,
                status
            FROM raw_orders
            WHERE status IS NOT NULL
        """)

defs = Definitions(
    assets=[raw_orders, cleaned_orders],
    resources={"duckdb": DuckDBResource(database="warehouse.db")}
)

Airbyte (Low-Code ELT)

Docker Setup

git clone https://github.com/airbytehq/airbyte.git
cd airbyte
./run-ab-platform.sh

API Configuration

import requests

# Create source
source = requests.post(
    "http://localhost:8000/api/v1/sources/create",
    json={
        "name": "My PostgreSQL",
        "sourceDefinitionId": "decd338e-5647-4c0b-adf4-da0e75f5a750",
        "workspaceId": "...",
        "connectionConfiguration": {
            "host": "localhost",
            "port": 5432,
            "database": "source_db"
        }
    }
)

Tool Selection Guide

Comparison Matrix

File Format Handling

Table of Contents

• CSV Processing
• Python (Polars - Recommended)
• TypeScript (papaparse)
• JSON Processing
• JSON Lines (Recommended for streaming)
• Nested JSON
• Parquet (Analytics Recommended)
• Python
• Rust (arrow-rs)
• Excel Processing
• Python (openpyxl + polars)
• Schema Validation
• Python with Pandera
• Format Selection Guide

CSV Processing

Python (Polars - Recommended)

import polars as pl

# Basic read with type inference
df = pl.read_csv("data.csv")

# With explicit schema
df = pl.read_csv(
    "data.csv",
    schema={
        "id": pl.Int64,
        "name": pl.Utf8,
        "amount": pl.Float64,
        "created_at": pl.Datetime
    },
    null_values=["", "NULL", "N/A"],
    skip_rows=1  # Skip header if needed
)

# Chunked reading for large files
reader = pl.read_csv_batched("large.csv", batch_size=100_000)
while True:
    batch = reader.next_batches(1)
    if not batch:
        break
    process_batch(batch[0])

TypeScript (papaparse)

import Papa from "papaparse";
import fs from "fs";

// Streaming parse
const file = fs.createReadStream("data.csv");

Papa.parse(file, {
  header: true,
  dynamicTyping: true,
  step: (row) => {
    // Process each row
    processRow(row.data);
  },
  complete: () => {
    console.log("Parsing complete");
  }
});

JSON Processing

JSON Lines (Recommended for streaming)

import polars as pl
import json

# Polars native NDJSON
df = pl.read_ndjson("events.jsonl")

# Streaming JSON Lines
def stream_jsonl(path: str):
    with open(path) as f:
        for line in f:
            yield json.loads(line)

# Write JSON Lines
df.write_ndjson("output.jsonl")

Nested JSON

# Flatten nested structures
df = pl.read_json("nested.json")
df = df.unnest("metadata").unnest("user")

Parquet (Analytics Recommended)

Python

import polars as pl

# Read with column selection (efficient!)
df = pl.read_parquet(
    "data.parquet",
    columns=["id", "amount", "created_at"]
)

# Read from S3 directly
df = pl.read_parquet("s3://bucket/data.parquet")

# Write with compression
df.write_parquet("output.parquet", compression="zstd")

# Partitioned writes
df.write_parquet(
    "output/",
    partition_by=["year", "month"]
)

Rust (arrow-rs)

use arrow::parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use std::fs::File;

fn read_parquet(path: &str) -> Result<Vec<RecordBatch>> {
    let file = File::open(path)?;
    let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
    let reader = builder.build()?;

    reader.collect()
}

Excel Processing

Python (openpyxl + polars)

import polars as pl

# Read specific sheet
df = pl.read_excel(
    "data.xlsx",
    sheet_name="Sheet1",
    read_options={"header_row": 0}
)

# Read all sheets
sheets = pl.read_excel("data.xlsx", sheet_name=None)
for name, df in sheets.items():
    print(f"Sheet: {name}, Rows: {len(df)}")

Schema Validation

Python with Pandera

import pandera as pa
import polars as pl

schema = pa.DataFrameSchema({
    "id": pa.Column(int, nullable=False, unique=True),
    "email": pa.Column(str, pa.Check.str_matches(r'^[\w\.-]+@[\w\.-]+\.\w+$')),
    "amount": pa.Column(float, pa.Check.ge(0)),
    "status": pa.Column(str, pa.Check.isin(["pending", "completed", "failed"]))
})

# Validate
df = pl.read_csv("data.csv")
validated = schema.validate(df.to_pandas())

Format Selection Guide

Streaming Data Ingestion

Table of Contents

• Apache Kafka
• Python (confluent-kafka)
• TypeScript (kafkajs)
• Rust (rdkafka)
• AWS Kinesis
• Python (boto3)
• Google Pub/Sub
• Python
• Exactly-Once Semantics
• Pattern: Idempotent Processing
• Pattern: Outbox for Reliability
• Dead Letter Queues
• Backpressure Handling

Apache Kafka

Python (confluent-kafka)

from confluent_kafka import Consumer, KafkaError
import json

def create_consumer(group_id: str, topics: list[str]) -> Consumer:
    conf = {
        'bootstrap.servers': 'localhost:9092',
        'group.id': group_id,
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False  # Manual commit for exactly-once
    }
    consumer = Consumer(conf)
    consumer.subscribe(topics)
    return consumer

def consume_messages(consumer: Consumer, batch_size: int = 100):
    """Consume messages with manual commit."""
    messages = []

    while True:
        msg = consumer.poll(timeout=1.0)

        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            raise Exception(msg.error())

        messages.append(json.loads(msg.value().decode()))

        if len(messages) >= batch_size:
            yield messages
            consumer.commit()
            messages = []

# Usage
consumer = create_consumer("my-group", ["events"])
for batch in consume_messages(consumer):
    process_batch(batch)

TypeScript (kafkajs)

import { Kafka, Consumer, EachBatchPayload } from "kafkajs";

const kafka = new Kafka({
  clientId: "my-app",
  brokers: ["localhost:9092"]
});

const consumer = kafka.consumer({ groupId: "my-group" });

async function startConsumer() {
  await consumer.connect();
  await consumer.subscribe({ topic: "events", fromBeginning: true });

  await consumer.run({
    eachBatch: async ({ batch, resolveOffset, heartbeat }: EachBatchPayload) => {
      for (const message of batch.messages) {
        const event = JSON.parse(message.value!.toString());
        await processEvent(event);
        resolveOffset(message.offset);
        await heartbeat();
      }
    }
  });
}

Rust (rdkafka)

use rdkafka::consumer::{Consumer, StreamConsumer};
use rdkafka::Message;

async fn consume_events(consumer: StreamConsumer) -> Result<()> {
    loop {
        match consumer.recv().await {
            Ok(msg) => {
                if let Some(payload) = msg.payload() {
                    let event: Event = serde_json::from_slice(payload)?;
                    process_event(event).await?;
                }
                consumer.commit_message(&msg, CommitMode::Async)?;
            }
            Err(e) => eprintln!("Kafka error: {}", e),
        }
    }
}

AWS Kinesis

Python (boto3)

import boto3
import json

kinesis = boto3.client('kinesis')

def consume_kinesis(stream_name: str, shard_id: str):
    # Get shard iterator
    response = kinesis.get_shard_iterator(
        StreamName=stream_name,
        ShardId=shard_id,
        ShardIteratorType='LATEST'
    )
    shard_iterator = response['ShardIterator']

    while True:
        response = kinesis.get_records(
            ShardIterator=shard_iterator,
            Limit=100
        )

        for record in response['Records']:
            data = json.loads(record['Data'])
            yield data

        shard_iterator = response['NextShardIterator']

Google Pub/Sub

Python

from google.cloud import pubsub_v1
import json

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path("project", "subscription")

def callback(message):
    data = json.loads(message.data.decode())
    process_message(data)
    message.ack()

streaming_pull = subscriber.subscribe(subscription_path, callback=callback)

# Run forever
streaming_pull.result()

Exactly-Once Semantics

Pattern: Idempotent Processing

async def process_with_idempotency(event: dict):
    event_id = event["id"]

    # Check if already processed
    existing = await db.execute(
        "SELECT 1 FROM processed_events WHERE event_id = ?",
        (event_id,)
    )
    if existing:
        return  # Skip duplicate

    # Process in transaction
    async with db.transaction():
        await process_event(event)
        await db.execute(
            "INSERT INTO processed_events (event_id, processed_at) VALUES (?, ?)",
            (event_id, datetime.utcnow())
        )

Pattern: Outbox for Reliability

async def process_with_outbox(event: dict):
    async with db.transaction():
        # 1. Write to outbox
        await db.execute(
            "INSERT INTO outbox (event_id, payload, status) VALUES (?, ?, 'pending')",
            (event["id"], json.dumps(event))
        )

        # 2. Process event
        result = await process_event(event)

        # 3. Mark complete
        await db.execute(
            "UPDATE outbox SET status = 'complete' WHERE event_id = ?",
            (event["id"],)
        )

    return result

Dead Letter Queues

async def consume_with_dlq(consumer, dlq_producer):
    for msg in consumer:
        try:
            await process_message(msg)
            consumer.commit()
        except Exception as e:
            # Send to DLQ after max retries
            if msg.retry_count >= 3:
                await dlq_producer.send(
                    topic="events-dlq",
                    value={
                        "original": msg.value,
                        "error": str(e),
                        "failed_at": datetime.utcnow().isoformat()
                    }
                )
                consumer.commit()
            else:
                # Re-queue for retry
                raise

Backpressure Handling

import asyncio
from asyncio import Semaphore

class BackpressureConsumer:
    def __init__(self, max_concurrent: int = 100):
        self.semaphore = Semaphore(max_concurrent)

    async def process_with_backpressure(self, message):
        async with self.semaphore:
            await process_message(message)

    async def consume(self, consumer):
        tasks = []
        async for msg in consumer:
            task = asyncio.create_task(self.process_with_backpressure(msg))
            tasks.append(task)

            # Periodically clean completed tasks
            if len(tasks) > 1000:
                tasks = [t for t in tasks if not t.done()]