
ml-data-pipeline-architecture
Patterns for efficient ML data pipelines using Polars, Arrow, and ClickHouse. TRIGGERS - data pipeline, polars vs pandas, arrow format
- Installs
- 129
- Last 24h
- —
- Trending rank
- —
- Security
- Not audited
Install
Works with every agent the skills CLI supports.
npx skills add https://github.com/terrylica/cc-skills --skill ml-data-pipeline-architecture- 01
Install
Run the command in your project. The CLI asks which agents to install for.
- 02
Describe the task
Ask your agent for something this skill covers. It loads ml-data-pipeline-architecture automatically when relevant.
- 03
Review the result
The agent follows the SKILL.md steps, scripts and references bundled with the skill.
SKILL.md
The instructions your agent reads when the skill is triggered.
ML Data Pipeline Architecture
Patterns for efficient ML data pipelines using Polars, Arrow, and ClickHouse.
ADR: 2026-01-22-polars-preference-hook (efficiency preferences framework)
Note: A PreToolUse hook enforces Polars preference. To use Pandas, add
# polars-exception: <reason>at file top.
Self-Evolving Skill: This skill improves through use. If instructions are wrong, parameters drifted, or a workaround was needed — fix this file immediately, don't defer. Only update for real, reproducible issues.
When to Use This Skill
Use this skill when:
- Deciding between Polars and Pandas for a data pipeline
- Optimizing memory usage with zero-copy Arrow patterns
- Loading data from ClickHouse into PyTorch DataLoaders
- Implementing lazy evaluation for large datasets
- Migrating existing Pandas code to Polars
1. Decision Tree: Polars vs Pandas
Dataset size?
├─ < 1M rows → Pandas OK (simpler API, richer ecosystem)
├─ 1M-10M rows → Consider Polars (2-5x faster, less memory)
└─ > 10M rows → Use Polars (required for memory efficiency)
Operations?
├─ Simple transforms → Either works
├─ Group-by aggregations → Polars 5-10x faster
├─ Complex joins → Polars with lazy evaluation
└─ Streaming/chunked → Polars scan_* functions
Integration?
├─ scikit-learn heavy → Pandas (better interop)
├─ PyTorch/custom → Polars + Arrow (zero-copy to tensor)
└─ ClickHouse source → Arrow stream → Polars (optimal)
2. Zero-Copy Pipeline Architecture
The Problem with Pandas
# BAD: 3 memory copies
df = pd.read_sql(query, conn) # Copy 1: DB → pandas
X = df[features].values # Copy 2: pandas → numpy
tensor = torch.from_numpy(X) # Copy 3: numpy → tensor
# Peak memory: 3x data size
The Solution with Arrow
# GOOD: 0-1 memory copies
import clickhouse_connect
import polars as pl
import torch
client = clickhouse_connect.get_client(...)
arrow_table = client.query_arrow("SELECT * FROM bars") # Arrow in DB memory
df = pl.from_arrow(arrow_table) # Zero-copy view
X = df.select(features).to_numpy() # Single allocation
tensor = torch.from_numpy(X) # View (no copy)
# Peak memory: 1.2x data size
3. ClickHouse Integration Patterns
Pattern A: Arrow Stream (Recommended)
def query_arrow(client, query: str) -> pl.DataFrame:
"""ClickHouse → Arrow → Polars (zero-copy chain)."""
arrow_table = client.query_arrow(f"{query} FORMAT ArrowStream")
return pl.from_arrow(arrow_table)
# Usage
df = query_arrow(client, "SELECT * FROM bars WHERE ts >= '2024-01-01'")
Pattern B: Polars Native (Simpler)
# Polars has native ClickHouse support (see pola.rs for version requirements)
df = pl.read_database_uri(
query="SELECT * FROM bars",
uri="clickhouse://user:pass@host/db"
)
Pattern C: Parquet Export (Batch Jobs)
# For reproducible batch processing
client.query("SELECT * FROM bars INTO OUTFILE 'data.parquet' FORMAT Parquet")
df = pl.scan_parquet("data.parquet") # Lazy, memory-mapped
4. PyTorch DataLoader Integration
Minimal Change Pattern
from torch.utils.data import TensorDataset, DataLoader
# Accept both pandas and polars
def prepare_data(df) -> tuple[torch.Tensor, torch.Tensor]:
if isinstance(df, pd.DataFrame):
df = pl.from_pandas(df)
X = df.select(features).to_numpy()
y = df.select(target).to_numpy()
return (
torch.from_numpy(X).float(),
torch.from_numpy(y).float()
)
X, y = prepare_data(df)
dataset = TensorDataset(X, y)
loader = DataLoader(dataset, batch_size=32, pin_memory=True)
Custom PolarsDataset (Large Data)
class PolarsDataset(torch.utils.data.Dataset):
"""Memory-efficient dataset from Polars DataFrame."""
def __init__(self, df: pl.DataFrame, features: list[str], target: str):
self.arrow = df.to_arrow() # Arrow backing for zero-copy slicing
self.features = features
self.target = target
def __len__(self) -> int:
return self.arrow.num_rows
def __getitem__(self, idx: int) -> tuple[torch.Tensor, torch.Tensor]:
row = self.arrow.slice(idx, 1)
x = torch.tensor([row[f][0].as_py() for f in self.features], dtype=torch.float32)
y = torch.tensor(row[self.target][0].as_py(), dtype=torch.float32)
return x, y
5. Lazy Evaluation Patterns
Pipeline Composition
# Define transformations lazily (no computation yet)
pipeline = (
pl.scan_parquet("raw_data.parquet")
.filter(pl.col("timestamp") >= start_date)
.with_columns([
(pl.col("close").pct_change()).alias("returns"),
(pl.col("volume").log()).alias("log_volume"),
])
.select(features + [target])
)
# Execute only when needed
train_df = pipeline.filter(pl.col("timestamp") < split_date).collect()
test_df = pipeline.filter(pl.col("timestamp") >= split_date).collect()
Streaming for Large Files
# Process file in chunks (never loads full file)
def process_large_file(path: str, chunk_size: int = 100_000):
reader = pl.scan_parquet(path)
for batch in reader.iter_batches(n_rows=chunk_size):
# Process each chunk
features = compute_features(batch)
yield features.to_numpy()
6. Schema Validation
Pydantic for Config
from pydantic import BaseModel, field_validator
class FeatureConfig(BaseModel):
features: list[str]
target: str
seq_len: int = 15
@field_validator("features")
@classmethod
def validate_features(cls, v):
required = {"returns_vs", "momentum_z", "atr_pct"}
missing = required - set(v)
if missing:
raise ValueError(f"Missing required features: {missing}")
return v
DataFrame Schema Validation
def validate_schema(df: pl.DataFrame, required: list[str], stage: str) -> None:
"""Fail-fast schema validation."""
missing = [c for c in required if c not in df.columns]
if missing:
raise ValueError(
f"[{stage}] Missing columns: {missing}\n"
f"Available: {sorted(df.columns)}"
)
7. Performance Benchmarks
| Operation | Pandas | Polars | Speedup |
|---|---|---|---|
| Read CSV (1GB) | 45s | 4s | 11x |
| Filter rows | 2.1s | 0.4s | 5x |
| Group-by agg | 3.8s | 0.3s | 13x |
| Sort | 5.2s | 0.4s | 13x |
| Memory peak | 10GB | 2.5GB | 4x |
Benchmark: 50M rows, 20 columns, MacBook M2
8. Migration Checklist
Phase 1: Add Arrow Support
- Add
polars = "<version>"to dependencies (see PyPI) - Implement
query_arrow()in data client - Verify zero-copy with memory profiler
Phase 2: Polars at Entry Points
- Add
pl.from_pandas()wrapper at trainer entry - Update
prepare_sequences()to accept both types - Add schema validation after conversion
Phase 3: Full Lazy Evaluation
- Convert file reads to
pl.scan_* - Compose transformations lazily
- Call
.collect()only before.to_numpy()
9. Anti-Patterns to Avoid
DON'T: Mix APIs Unnecessarily
# BAD: Convert back and forth
df_polars = pl.from_pandas(df_pandas)
df_pandas_again = df_polars.to_pandas() # Why?
DON'T: Collect Too Early
# BAD: Defeats lazy evaluation
df = pl.scan_parquet("data.parquet").collect() # Full load
filtered = df.filter(...) # After the fact
# GOOD: Filter before collect
df = pl.scan_parquet("data.parquet").filter(...).collect()
DON'T: Ignore Memory Pressure
# BAD: Loads entire file
df = pl.read_parquet("huge_file.parquet")
# GOOD: Stream in chunks
for batch in pl.scan_parquet("huge_file.parquet").iter_batches():
process(batch)
References
- Polars User Guide
- Polars Migration Guide
- Apache Arrow Python
- ClickHouse Python Client
- PyTorch Data Loading
- Polars Preference Hook ADR
Troubleshooting
| Issue | Cause | Solution |
|---|---|---|
| Memory spike during load | Collecting too early | Use lazy evaluation, call collect() only when needed |
| Arrow conversion fails | Unsupported data type | Check for object columns, convert to native types |
| ClickHouse connection error | Wrong port or credentials | Verify host:8123 (HTTP) or host:9000 (native) |
| Zero-copy not working | Intermediate pandas conversion | Remove to_pandas() calls, stay in Arrow/Polars |
| Polars hook blocking code | Pandas used without exception | Add # polars-exception: reason comment at file top |
| Slow group-by operations | Using pandas for large datasets | Migrate to Polars for 5-10x speedup |
| Schema validation failure | Column names case-sensitive | Verify exact column names from source |
| PyTorch DataLoader OOM | Loading full dataset into memory | Use PolarsDataset with Arrow backing for lazy access |
| Parquet scan performance | Not using predicate pushdown | Add filters before collect() for lazy evaluation |
| Type mismatch in tensor | Float64 vs Float32 mismatch | Explicitly cast with .cast(pl.Float32) before numpy |
Post-Execution Reflection
After this skill completes, reflect before closing the task:
- Locate yourself. — Find this SKILL.md's canonical path before editing.
- What failed? — Fix the instruction that caused it.
- What worked better than expected? — Promote to recommended practice.
- What drifted? — Fix any script, reference, or dependency that no longer matches reality.
- Log it. — Evolution-log entry with trigger, fix, and evidence.
Do NOT defer. The next invocation inherits whatever you leave behind.
FAQ
What does the ml-data-pipeline-architecture skill do?
Patterns for efficient ML data pipelines using Polars, Arrow, and ClickHouse. TRIGGERS - data pipeline, polars vs pandas, arrow format
How do I install ml-data-pipeline-architecture?
Run npx skills add https://github.com/terrylica/cc-skills --skill ml-data-pipeline-architecture in your project. The skills CLI detects the agents you use and copies the skill into the right folder. For Claude Code you can also place the skill folder at ~/.claude/skills/ml-data-pipeline-architecture/ (personal) or .claude/skills/ml-data-pipeline-architecture/ (project).
Which AI agents can use ml-data-pipeline-architecture?
Any agent that supports the open Agent Skills format (a folder with a SKILL.md file), including Claude Code, Codex, Cursor, GitHub Copilot, Gemini CLI, Windsurf, Cline, OpenCode and more.
Is ml-data-pipeline-architecture safe to install?
No partner security audit is available yet. Skills can run commands and scripts, so read the SKILL.md and any bundled files before installing.
Who maintains ml-data-pipeline-architecture?
It is published by terrylica in terrylica/cc-skills. It has 129 recorded installs.
More from terrylica/cc-skills
mise-tasks
terrylica/cc-skills
Orchestrate multi-step project workflows using mise task definitions with dependency management and argument handling. Use whenever the user.
pandoc-pdf-generation
terrylica/cc-skills
PDF generation from markdown via Pandoc/XeLaTeX. TRIGGERS - markdown for PDF, print document, pandoc
mql5-indicator-patterns
terrylica/cc-skills
MQL5 indicator development patterns. TRIGGERS - MQL5 indicator, OnCalculate, indicator buffers, MetaTrader 5.
python-logging-best-practices
terrylica/cc-skills
Python logging with loguru, structlog, and orjson. TRIGGERS - loguru, structlog, structured logging
gdrive-access
terrylica/cc-skills
Access Google Drive via CLI with 1Password OAuth: list files, download, and sync folders. TRIGGERS - google drive, gdrive, drive folder, download drive, sync drive.
Popular data & databases skills
supabase-postgres-best-practices
supabase/agent-skills
Postgres best practices maintained by Supabase, for Postgres running anywhere. Load this skill BEFORE writing or changing anything that lives in a Postgres database: creating or altering tables and columns (including choosing column types), schema design, migrations and declarative schema files, RLS policies and the tests that verify them, indexes, triggers, database functions, queues and scheduled jobs (pgcron, pgmq), vector/semantic search (pgvector), and restoring dumps (pgrestore) or importing data. Also load it when diagnosing slow queries, high CPU, timeouts, EXPLAIN plans, connection exhaustion, locking, bloat, or rows visible to the wrong user or tenant. This is not just a performance guide — schema, migration, security, and SQL authoring tasks need these rules too, even for a one-column change or a single query.
prisma-client-api
prisma/skills
Prisma Client API reference covering model queries, filters, operators, and client methods. Use when writing database queries, using CRUD operations, filtering data, or configuring Prisma Client. Triggers on "prisma query", "findMany", "create", "update", "delete", "$transaction".
prisma-database-setup
prisma/skills
Deprecated compatibility name for prisma-orm-setup. Use only when a user or existing integration explicitly invokes prisma-database-setup; otherwise select prisma-orm-setup.
prisma-cli
prisma/skills
Prisma ORM 7 CLI reference for existing Prisma 7 applications or explicitly selected Prisma 7 workflows. Covers prisma init, generate, migrate, db, dev, studio, and mcp. For setup, connection repair, or Prisma 8, use prisma-orm-setup; not the Prisma Platform CLI.
prisma-postgres
prisma/skills
Deprecated compatibility name for prisma-postgres-setup. Use only when a user or existing integration explicitly invokes prisma-postgres; otherwise select prisma-postgres-setup.
prisma-driver-adapter-implementation
prisma/skills
Required reference for Prisma ORM 7 SQL driver adapter work. Use when implementing or modifying adapters, adding database drivers, or touching SqlDriverAdapter, Transaction, savepoint, result mapping, or DriverAdapterError behavior. Covers current transaction lifecycle, optional savepoint hooks, original database-error preservation, and verification.