Data Pipeline Patterns Agent Skill

by vibeeval

ETL/ELT patterns, batch vs streaming, idempotency, data quality framework, and pipeline orchestration

807 tokens

context cost

the whole folder, loaded on every use

files

instructions only

copies elsewhere

how many repositories repackaged it

521

stars on the repo

on the repository, not the skill itself

Install

one command, takes just this skill from the repository

npx skills add https://github.com/vibeeval/vibecosystem --skill data-pipeline-patterns

The instruction itself

9 sections, as written by the author

Data Pipeline Patterns

ETL vs ELT Decision

| Kriter | ETL | ELT |

|--------|-----|-----|

| Transform location | Pipeline'da | Data warehouse'da |

| Data volume | Küçük-orta | Büyük |

| Flexibility | Düşük | Yüksek |

| Cost | Compute-heavy | Storage-heavy |

| Use case | Legacy, compliance | Modern analytics |

Batch vs Streaming

| Kriter | Batch | Streaming |

|--------|-------|-----------|

| Latency | Dakika-saat | Saniye-milisaniye |

| Complexity | Düşük | Yüksek |

| Cost | Düşük | Yüksek |

| Use case | Reporting, ETL | Real-time alerts, dashboards |

| Tool | Airflow, dbt | Kafka Streams, Flink |

Idempotency Patterns

# Pattern 1: Upsert
INSERT INTO target (id, name, updated_at)
VALUES (%(id)s, %(name)s, %(ts)s)
ON CONFLICT (id) DO UPDATE SET
  name = EXCLUDED.name,
  updated_at = EXCLUDED.updated_at

# Pattern 2: Partition overwrite
DELETE FROM target WHERE partition_date = '2026-03-14';
INSERT INTO target SELECT * FROM staging WHERE partition_date = '2026-03-14';

# Pattern 3: Checkpoint
last_checkpoint = get_checkpoint('pipeline_x')
new_data = source.query(f"WHERE updated_at > '{last_checkpoint}'")
process(new_data)
save_checkpoint('pipeline_x', max(new_data.updated_at))

Data Quality Framework

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int, pa.Check.gt(0), nullable=False),
    "email": pa.Column(str, pa.Check.str_matches(r'^.+@.+\..+$')),
    "age": pa.Column(int, pa.Check.in_range(0, 150), nullable=True),
    "created_at": pa.Column(pa.DateTime, pa.Check.less_than_or_equal_to(pd.Timestamp.now()))
})

validated_df = schema.validate(df)  # Fail on invalid data

Quality Dimensions

| Dimension | Kontrol | Tool |

|-----------|---------|------|

| Completeness | NULL ratio < threshold | Great Expectations |

| Accuracy | Value range checks | pandera |

| Freshness | Last update < SLA | Airflow sensor |

| Uniqueness | Duplicate check | SQL DISTINCT |

| Consistency | Cross-table referential integrity | dbt test |

Pipeline Orchestration

# Airflow DAG
from airflow import DAG
from airflow.operators.python import PythonOperator

with DAG('daily_etl', schedule='0 6 * * *', catchup=False) as dag:
    extract = PythonOperator(task_id='extract', python_callable=extract_fn)
    transform = PythonOperator(task_id='transform', python_callable=transform_fn)
    load = PythonOperator(task_id='load', python_callable=load_fn)
    validate = PythonOperator(task_id='validate', python_callable=validate_fn)

    extract >> transform >> load >> validate

Checklist

[ ] Pipeline idempotent (rerun safe)
[ ] Data quality checks her adımda
[ ] Dead letter queue (failed records)
[ ] Monitoring + alerting aktif
[ ] Schema evolution handled
[ ] Backfill mekanizması var
[ ] Retry logic (exponential backoff)
[ ] Data lineage tracked

Anti-Patterns

Pipeline'da hardcoded credentials
Idempotent olmayan transform
Data quality check'siz load
Monolithic pipeline (parçala)
Silent failure (error swallowing)

Other skills for the same job

different authors, same section of the catalogue

Modal

by christophacham

×3

Run Python code in the cloud with serverless containers, GPUs, and autoscaling. Use when deploying ML models, running batch processing jobs, scheduling compute-intensive tasks, or serving APIs that require GPU acceleration or dynamic scaling.

17k tokens

Github Workflow Automation

by ComeOnOliver

×3

Advanced GitHub Actions workflow automation with AI swarm coordination, intelligent CI/CD pipelines, and comprehensive repository management

Google Cloud Platform CLI - manage GCP resources including Compute Engine, Cloud Run, GKE, Cloud Functions, Storage, BigQuery, and more.

Expert backend architect specializing in scalable API design, microservices architecture, and distributed systems. Masters REST/GraphQL/gRPC APIs, event-driven architectures, service mesh patterns, and modern backend frameworks. Handles service boundary definition, inter-service communication, resilience patterns, and observability. Use PROACTIVELY when creating new backend services or APIs.

Aspire skill covering the Aspire CLI, AppHost orchestration, service discovery, integrations, MCP server, VS Code extension, Dev Containers, GitHub Codespaces, templates, dashboard, and deployment. Use when the user asks to create, run, debug, configure, deploy, or troubleshoot an Aspire distributed application.

21k tokens

Bigquery Pipeline Audit

by github

vendor ×1

Audits Python + BigQuery pipelines for cost safety, idempotency, and production readiness. Returns a structured report with exact patch locations.

Microsoft Store Developer CLI (msstore) for publishing Windows applications to the Microsoft Store. Use when asked to configure Store credentials, list Store apps, check submission status, publish submissions, manage package flights, set up CI/CD for Store publishing, or integrate with Partner Center. Supports Windows App SDK/WinUI, UWP, .NET MAUI, Flutter, Electron, React Native, and PWA applications.

4k tokens

How to use it

Copy the folder

Take vibeeval/data-pipeline-patterns from the repository into ~/.claude/skills for personal use, or into .claude/skills inside a project.

Check the name does not clash

The agent identifies a skill by the name field in its header. Two skills with the same name cannot sit side by side — one of them will be ignored.