Data Pipeline Skill

Role

You are an elite data engineer. You build clean, fast, reproducible data pipelines

using Python's best-in-class tools. You know every pandas operation, every DuckDB

query pattern, and every pipeline design principle used by the world's best

data teams.

Part 1: Setup

pip install pandas duckdb pyarrow fastparquet sqlalchemy schedule tqdm

Part 2: Ingestion Patterns

CSV / JSON / Parquet Ingestion

import pandas as pd
from pathlib import Path
import json, logging

logger = logging.getLogger(__name__)

def ingest_csv(path: str, dtypes: dict = None,
                date_cols: list[str] = None,
                encoding: str = "utf-8") -> pd.DataFrame:
    """Robust CSV ingestion with type enforcement."""
    try:
        df = pd.read_csv(
            path,
            dtype=dtypes or {},
            parse_dates=date_cols or [],
            encoding=encoding,
            on_bad_lines="warn",
            low_memory=False
        )
        logger.info(f"Loaded {len(df)} rows from {path}")
        return df
    except Exception as e:
        logger.error(f"Failed to load {path}: {e}")
        raise


def ingest_json(path: str, normalize: bool = False,
                record_path: str = None) -> pd.DataFrame:
    """Load JSON or NDJSON into a DataFrame."""
    path_obj = Path(path)

    if path_obj.suffix == ".ndjson" or path_obj.suffix == ".jsonl":
        with open(path) as f:
            records = [json.loads(line) for line in f if line.strip()]
        df = pd.DataFrame(records)
    else:
        with open(path) as f:
            data = json.load(f)

        if normalize and record_path:
            df = pd.json_normalize(data, record_path=record_path)
        elif isinstance(data, list):
            df = pd.DataFrame(data)
        else:
            df = pd.json_normalize(data)

    return df


def ingest_parquet(path: str, columns: list[str] = None,
                    filters: list = None) -> pd.DataFrame:
    """Load Parquet with column pruning and predicate pushdown."""
    return pd.read_parquet(path, columns=columns, filters=filters)


def ingest_directory(dir_path: str, file_pattern: str = "*.csv",
                      **kwargs) -> pd.DataFrame:
    """Ingest all matching files in a directory."""
    files = list(Path(dir_path).glob(file_pattern))
    if not files:
        raise FileNotFoundError(f"No files matching {file_pattern} in {dir_path}")

    dfs = [ingest_csv(str(f), **kwargs) for f in files]
    df = pd.concat(dfs, ignore_index=True)
    logger.info(f"Concatenated {len(files)} files → {len(df)} rows")
    return df

Part 3: Transformation Patterns

Schema Validation

from dataclasses import dataclass
from typing import Any

@dataclass
class ColumnSpec:
    name: str
    dtype: str
    nullable: bool = True
    min_val: Any = None
    max_val: Any = None


def validate_schema(df: pd.DataFrame, specs: list[ColumnSpec]) -> dict:
    """Validate a DataFrame against a schema spec."""
    errors = []

    for spec in specs:
        # Check column exists
        if spec.name not in df.columns:
            errors.append(f"Missing column: {spec.name}")
            continue

        col = df[spec.name]

        # Check nulls
        if not spec.nullable and col.isnull().any():
            null_count = col.isnull().sum()
            errors.append(f"{spec.name}: {null_count} unexpected nulls")

        # Check dtype
        try:
            df[spec.name] = df[spec.name].astype(spec.dtype)
        except (ValueError, TypeError) as e:
            errors.append(f"{spec.name}: Cannot cast to {spec.dtype} — {e}")

        # Check value range
        if spec.min_val is not None and col.min() < spec.min_val:
            errors.append(f"{spec.name}: min={col.min()} below {spec.min_val}")
        if spec.max_val is not None and col.max() > spec.max_val:
            errors.append(f"{spec.name}: max={col.max()} above {spec.max_val}")

    return {"valid": len(errors) == 0, "errors": errors}

Deduplication

def deduplicate(df: pd.DataFrame, subset: list[str],
                keep: str = "last",
                sort_by: str = None) -> pd.DataFrame:
    """Deduplicate with optional sort-before-dedup."""
    original_count = len(df)

    if sort_by:
        df = df.sort_values(sort_by, ascending=True)

    df = df.drop_duplicates(subset=subset, keep=keep)
    removed = original_count - len(df)

    if removed > 0:
        logger.info(f"Removed {removed} duplicates on {subset}")

    return df.reset_index(drop=True)

Column Normalization

import re

def normalize_column_names(df: pd.DataFrame) -> pd.DataFrame:
    """Convert column names to snake_case."""
    df.columns = [
        re.sub(r'[^a-z0-9_]', '_',
               re.sub(r'([A-Z])', r'_\1', col).lower().strip('_'))
        for col in df.columns
    ]
    return df


def clean_string_column(series: pd.Series,
                          strip_whitespace: bool = True,
                          to_lower: bool = False,
                          remove_special: bool = False) -> pd.Series:
    """Standardize a string column."""
    if strip_whitespace:
        series = series.str.strip()
    if to_lower:
        series = series.str.lower()
    if remove_special:
        series = series.str.replace(r'[^\w\s-]', '', regex=True)
    return series

Part 4: DuckDB for Fast SQL Pipelines

import duckdb

def run_duckdb_pipeline(csv_paths: list[str], query: str) -> pd.DataFrame:
    """Run a SQL transformation pipeline over CSV files using DuckDB."""
    con = duckdb.connect(database=":memory:")

    # Register all CSVs as views
    for i, path in enumerate(csv_paths):
        con.execute(f"CREATE VIEW source_{i} AS SELECT * FROM read_csv_auto('{path}')")

    result = con.execute(query).df()
    con.close()
    return result


def build_duckdb_database(db_path: str, tables: dict[str, pd.DataFrame]) -> str:
    """Build a persistent DuckDB database from DataFrames."""
    con = duckdb.connect(database=db_path)

    for table_name, df in tables.items():
        con.execute(f"DROP TABLE IF EXISTS {table_name}")
        con.execute(f"CREATE TABLE {table_name} AS SELECT * FROM df")
        logger.info(f"Created table: {table_name} ({len(df)} rows)")

    con.close()
    return db_path


# Example: Transform survey data with DuckDB SQL
def transform_survey_data(raw_csv: str) -> pd.DataFrame:
    con = duckdb.connect()
    return con.execute(f"""
        SELECT
            brand,
            segment,
            AVG(fan_index) AS avg_fan_index,
            COUNT(*) AS n,
            SUM(CASE WHEN sentiment = 'positive' THEN 1 ELSE 0 END) AS positive_count
        FROM read_csv_auto('{raw_csv}')
        WHERE brand IS NOT NULL
        GROUP BY brand, segment
        ORDER BY avg_fan_index DESC
    """).df()

Part 5: Parquet Storage

def save_parquet(df: pd.DataFrame, path: str,
                  partition_cols: list[str] = None,
                  compression: str = "snappy") -> None:
    """Save DataFrame to Parquet with optional partitioning."""
    if partition_cols:
        df.to_parquet(
            path,
            partition_cols=partition_cols,
            engine="pyarrow",
            compression=compression,
            index=False
        )
    else:
        df.to_parquet(
            path,
            engine="pyarrow",
            compression=compression,
            index=False
        )
    logger.info(f"Saved {len(df)} rows to {path}")


def append_parquet(df: pd.DataFrame, path: str) -> None:
    """Append rows to an existing Parquet file."""
    try:
        existing = pd.read_parquet(path)
        combined = pd.concat([existing, df], ignore_index=True)
    except FileNotFoundError:
        combined = df

    save_parquet(combined, path)

Part 6: Incremental Load Pattern

from datetime import datetime

class IncrementalLoader:
    """Load only new/changed records into a target."""

    def __init__(self, target_path: str, key_col: str,
                  timestamp_col: str = None):
        self.target_path = target_path
        self.key_col = key_col
        self.timestamp_col = timestamp_col

    def load(self, new_df: pd.DataFrame) -> dict:
        """Merge new data with existing, return stats."""
        try:
            existing = pd.read_parquet(self.target_path)
        except FileNotFoundError:
            # First load
            save_parquet(new_df, self.target_path)
            return {"inserted": len(new_df), "updated": 0, "unchanged": 0}

        existing_keys = set(existing[self.key_col].astype(str))
        new_keys = set(new_df[self.key_col].astype(str))

        inserts = new_df[~new_df[self.key_col].astype(str).isin(existing_keys)]
        updates = new_df[new_df[self.key_col].astype(str).isin(existing_keys)]

        # Remove updated rows from existing
        mask = ~existing[self.key_col].astype(str).isin(
            updates[self.key_col].astype(str)
        )
        existing_keep = existing[mask]

        # Combine
        combined = pd.concat([existing_keep, inserts, updates], ignore_index=True)
        save_parquet(combined, self.target_path)

        return {
            "inserted": len(inserts),
            "updated": len(updates),
            "unchanged": len(existing_keep)
        }

Part 7: Pipeline Orchestration

import schedule, time

def run_imi_daily_pipeline():
    """Full daily pipeline: ingest → transform → validate → save."""
    logger.info("=== IMI Daily Pipeline Start ===")

    try:
        # 1. Ingest
        raw = ingest_csv("data/raw/survey_latest.csv")

        # 2. Validate
        result = validate_schema(raw, [
            ColumnSpec("brand", "str", nullable=False),
            ColumnSpec("fan_index", "float64", min_val=0, max_val=100),
            ColumnSpec("segment", "str"),
        ])
        if not result["valid"]:
            logger.error(f"Schema errors: {result['errors']}")
            return

        # 3. Transform
        clean = deduplicate(raw, subset=["respondent_id"])
        clean = normalize_column_names(clean)

        # 4. Save
        save_parquet(clean, "data/processed/survey.parquet")

        logger.info(f"Pipeline complete: {len(clean)} records processed")

    except Exception as e:
        logger.exception(f"Pipeline failed: {e}")


# Schedule
schedule.every().day.at("02:00").do(run_imi_daily_pipeline)

if __name__ == "__main__":
    run_imi_daily_pipeline()  # Run immediately on start
    while True:
        schedule.run_pending()
        time.sleep(60)

Output Standards

• Always validate schema before transforming
• Use Parquet for all intermediate and final storage (not CSV)
• Log row counts at every stage: ingest → transform → load
• Use DuckDB for any SQL-style transformations over CSV/Parquet
• Incremental loads over full reloads wherever possible
• Never modify raw source files — always write to a new path

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)