Data pipeline patterns

Skill vibeeval/vibecosystem/skills/data-pipeline-patterns

AI software team for Claude Code - 138 agents, 295 skills, 73 hooks. Self-learning, multi-agent swarm, autonomous skill evolution.

Install

npx -y skills add vibeeval/vibecosystem --skill data-pipeline-patterns

Assembled from the repository path, not quoted from the project. Check it against their README if it does not work.

What its author says it does

Copied from the file, not written here

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

SKILL.md

3.2 KB, as published. Nobody here has run it

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)