data-pipeline-processor

name	data-pipeline-processor
description	Process data files through transformation pipelines with validation, cleaning, and export. Use for CSV/Excel/JSON data processing, encoding handling, batch operations, and data transformation workflows.
type	reference
version	1.1.0
category	development
related_skills	["yaml-workflow-executor","engineering-report-generator","parallel-file-processor"]
capabilities	[]
requires	[]
tags	[]
freedom	medium

Data Pipeline Processor

Quick Start

import pandas as pd
from pathlib import Path

# Simple pipeline: Load -> Transform -> Export
df = pd.read_csv("data/raw/source.csv")

# Transform
df = df[df['value'] > 0]  # Filter
df['date'] = pd.to_datetime(df['date'])  # Convert types
df = df.sort_values('date')  # Sort

# Export
Path("data/processed").mkdir(parents=True, exist_ok=True)
df.to_csv("data/processed/cleaned.csv", index=False)

print(f"Processed {len(df)} rows")

When to Use

• Processing CSV/Excel/JSON files with validation
• Data cleaning and transformation workflows
• Batch file processing with aggregation
• Handling encoding issues (UTF-8, Latin-1 fallback)
• ETL (Extract, Transform, Load) operations
• Data quality checks and reporting

Core Pattern

Input (CSV/Excel/JSON) -> Validate -> Transform -> Analyze -> Export

Implementation

Data Reader with Encoding Detection

import pandas as pd
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
import logging
import chardet

logger = logging.getLogger(__name__)



*See sub-skills  full details.*


```python
 dataclasses  dataclass, field
 typing  , , , 



 :
    
    is_valid: 
    errors: [] = field(default_factory=)

*See sub-skills  full details.*


```python
 :
    

     ():
        .df = df.copy()

     () -> :
        
        .df = .df.rename(columns=mapping)

*See sub-skills  full details.*


```python
 :
    


     () -> :
        
        Path(path).parent.mkdir(parents=, exist_ok=)
        df.to_csv(path, index=, **kwargs)
         path

*See sub-skills  full details.*


```python
 dataclasses  dataclass
 typing  , , , 



 :
    
    input_path: 
    output_path: 

*See sub-skills  full details.*





```yaml


:
  path: data/raw/source.csv
  options:
    delimiter: 
    skiprows: 

validation:

*See sub-skills  full details.*


```yaml


:
  path: data/processed/daily_data.csv

validation:
  required_columns:
    - date
    - category

*See sub-skills  full details.*



- [yaml-workflow-executor](../yaml-workflow-executor/SKILL.md) - Workflow orchestration
- [engineering-report-generator](../engineering-report-generator/SKILL.md) - Report generation
- [parallel-file-processor](../parallel-file-processor/SKILL.md) - Parallel file operations

---



- **** (-01-02): Upgraded to SKILL_TEMPLATE_v2   Quick Start, Error Handling, Metrics, Execution Checklist, additional examples
- **** (--): Initial release  DataReader, DataValidator, DataTransformer, pipeline orchestration



- [Example : Simple CSV Processing (+)](example--simple-csv-processing/SKILL.md)
- [Do (+)](do/SKILL.md)



- [Error Handling](error-handling/SKILL.md)
- [Execution Checklist](execution-checklist/SKILL.md)
- [Metrics](metrics/SKILL.md)