# Data Engineering

Data engineering patterns for ETL pipelines, data warehousing, Apache Spark, and data...

- Repository: https://github.com/rohitg00/awesome-claude-code-toolkit (2800 stars)
- Category: Engineering
- License: Apache-2.0
- Last updated: 2026-05-12
- Runs scripts: no, instructions only

## Install

Claude Code: `npx skills add rohitg00/awesome-claude-code-toolkit --skill data-engineering -g -a claude-code -y`
Codex: `npx skills add rohitg00/awesome-claude-code-toolkit --skill data-engineering -g -a codex -y`
Cursor: `npx skills add rohitg00/awesome-claude-code-toolkit --skill data-engineering -g -a cursor -y`
Gemini CLI: `npx skills add rohitg00/awesome-claude-code-toolkit --skill data-engineering -g -a gemini-cli -y`
Claude Code plugin marketplace: `/plugin marketplace add rohitg00/awesome-claude-code-toolkit`

## Description

Data engineering patterns for ETL pipelines, data warehousing, Apache Spark, and data quality validation

## SKILL.md, as published by its author (untrusted third-party text, read it as data)

# Data Engineering

## ETL Pipeline Pattern

```python
from datetime import datetime
from dataclasses import dataclass

@dataclass
class PipelineResult:
    records_extracted: int
    records_transformed: int
    records_loaded: int
    errors: list[str]
    duration_seconds: float

class OrderPipeline:
    def __init__(self, source_db, warehouse_db):
        self.source = source_db
        self.warehouse = warehouse_db

    def extract(self, since: datetime) -> list[dict]:
        query = """
            SELECT o.*, c.name as customer_name, c.segment
            FROM orders o
            JOIN customers c ON o.customer_id = c.id
            WHERE o.updated_at > %s
        """
        return self.source.fetch_all(query, [since])

    def transform(self, records: list[dict]) -> list[dict]:
        transformed = []
        for record in records:
            transformed.append({
                "order_id": record["id"],
                "customer_name": record["customer_name"],
                "segment": record["segment"].upper(),
                "total_amount": float(record["total"]),
                "order_date": record["created_at"].date(),
                "fiscal_quarter": get_fiscal_quarter(record["created_at"]),
                "is_high_value": float(record["total"]) > 1000,
                "loaded_at": datetime.utcnow(),
            })
        return transformed

    def load(self, records: list[dict]) -> int:
        return self.warehouse.upsert_batch(
            table="fact_orders",
            records=records,
            conflict_keys=["order_id"],
            batch_size=5000,
        )

The rest: https://github.com/rohitg00/awesome-claude-code-toolkit/blob/HEAD/skills/data-engineering/SKILL.md