Name: Data Engineering Data Pipeline
Author: sickn33

搵技能.../

# Batch ingestion with validation
from batch_ingestion import BatchDataIngester
from storage.delta_lake_manager import DeltaLakeManager
from data_quality.expectations_suite import DataQualityFramework

ingester = BatchDataIngester(config={})

# Extract with incremental loading
df = ingester.extract_from_database(
    connection_string='postgresql://host:5432/db',
    query='SELECT * FROM orders',
    watermark_column='updated_at',
    last_watermark=last_run_timestamp
)

# Validate
schema = {'required_fields': ['id', 'user_id'], 'dtypes': {'id': 'int64'}}
df = ingester.validate_and_clean(df, schema)

# Data quality checks
dq = DataQualityFramework()
result = dq.validate_dataframe(df, suite_name='orders_suite', data_asset_name='orders')

# Write to Delta Lake
delta_mgr = DeltaLakeManager(storage_path='s3://lake')
delta_mgr.create_or_update_table(
    df=df,
    table_name='orders',
    partition_columns=['order_date'],
    mode='append'
)

# Save failed records
ingester.save_dead_letter_queue('s3://lake/dlq/orders')

Data Engineering Data Pipeline | Skills Pool

Data Engineering Data Pipeline

Data Engineering Data Pipeline

Data Pipeline Architecture

Use this skill when

Do not use this skill when

Requirements

Core Capabilities

Instructions

1. Architecture Design

2. Ingestion Implementation

3. Orchestration

4. Transformation with dbt

5. Data Quality Framework

6. Storage Strategy

7. Monitoring & Cost Optimization

Example: Minimal Batch Pipeline

Output Deliverables

1. Architecture Documentation

2. Implementation Code

3. Configuration Files

4. Monitoring & Observability

5. Operations Guide

Success Criteria

Limitations

Clickhouse Io

Clickhouse Io

Claude Devfleet

Clickhouse Io

Ai First Engineering

Postgres Patterns