system-prompts-and-models-of-ai-tools/claude_skills/data-pipeline/SKILL.md

name	description	allowed-tools
data-pipeline	Expert in building ETL/ELT pipelines, data processing, transformation, and orchestration using tools like Airflow, Spark, and dbt. Use for data engineering tasks, building data workflows, or implementing data processing systems.	Read, Write, Edit, Grep, Glob, Bash

Data Pipeline Expert

Purpose

Build robust ETL/ELT pipelines for data processing, transformation, and orchestration.

Tools & Technologies

• Orchestration: Apache Airflow, Prefect, Dagster
• Processing: Apache Spark, dbt, Pandas
• Storage: S3, GCS, Data Lakes
• Warehouses: Snowflake, BigQuery, Redshift
• Streaming: Apache Kafka, AWS Kinesis
• Quality: Great Expectations, dbt tests

Airflow DAG Example

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': True,
}

with DAG(
    'user_analytics_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['analytics', 'users'],
) as dag:

    extract_users = PythonOperator(
        task_id='extract_users',
        python_callable=extract_from_api,
        op_kwargs={'endpoint': 'users'}
    )

    transform_data = PythonOperator(
        task_id='transform_data',
        python_callable=transform_user_data,
    )

    load_to_warehouse = PostgresOperator(
        task_id='load_to_warehouse',
        postgres_conn_id='warehouse',
        sql='sql/load_users.sql',
    )

    data_quality_check = PythonOperator(
        task_id='data_quality_check',
        python_callable=run_quality_checks,
    )

    extract_users >> transform_data >> load_to_warehouse >> data_quality_check

dbt Transformation

-- models/staging/stg_users.sql
with source as (
    select * from {{ source('raw', 'users') }}
),

transformed as (
    select
        id as user_id,
        lower(email) as email,
        created_at,
        updated_at,
        case
            when status = 'active' then true
            else false
        end as is_active
    from source
    where created_at is not null
)

select * from transformed

-- models/marts/fct_user_activity.sql
with user_events as (
    select * from {{ ref('stg_events') }}
),

aggregated as (
    select
        user_id,
        count(*) as total_events,
        count(distinct date(created_at)) as active_days,
        min(created_at) as first_event_at,
        max(created_at) as last_event_at
    from user_events
    group by 1
)

select * from aggregated

Success Criteria

• ✓ Data freshness < 1 hour
• ✓ Pipeline success rate > 99%
• ✓ Data quality checks passing
• ✓ Idempotent operations
• ✓ Monitoring and alerting