python
import pandas as pd

# Read data
df = pd.read_csv("sales.csv")

# Transform
df['revenue'] = df['price'] * df['quantity']
df['month'] = pd.to_datetime(df['date']).dt.month

# Aggregate
monthly = (df.groupby('month')
  .agg(total_revenue=('revenue', 'sum'),
       avg_order=('revenue', 'mean'),
       num_orders=('id', 'count'))
  .reset_index())

python
import polars as pl

# Lazy evaluation — builds a query plan
result = (
    pl.scan_csv("sales.csv")  # Lazy!
    .with_columns([
        (pl.col("price") * pl.col("quantity")).alias("revenue"),
        pl.col("date").str.to_datetime().dt.month().alias("month"),
    ])
    .group_by("month")
    .agg([
        pl.col("revenue").sum().alias("total_revenue"),
        pl.col("revenue").mean().alias("avg_order"),
        pl.count().alias("num_orders"),
    ])
    .sort("month")
    .collect()  # Execute the optimized plan
)

python
from datetime import datetime
import polars as pl
from sqlalchemy import create_engine

class ETLPipeline:
    def __init__(self, source_db: str, target_db: str):
        self.source = create_engine(source_db)
        self.target = create_engine(target_db)
    
    def extract(self, query: str) -> pl.DataFrame:
        """Extract data from source."""
        df = pl.read_database(query, self.source)
        print(f"Extracted {len(df)} rows")
        return df
    
    def transform(self, df: pl.DataFrame) -> pl.DataFrame:
        """Apply business transformations."""
        return (df
            .filter(pl.col("status") == "completed")
            .with_columns([
                (pl.col("amount") * 1.18).alias("amount_with_tax"),
                pl.col("created_at").dt.date().alias("date"),
                pl.when(pl.col("amount") > 1000)
                  .then(pl.lit("high"))
                  .otherwise(pl.lit("normal"))
                  .alias("tier"),
            ])
            .drop_nulls()
        )
    
    def load(self, df: pl.DataFrame, table: str):
        """Load into target database."""
        df.write_database(table, self.target, if_table_exists="append")
        print(f"Loaded {len(df)} rows into {table}")
    
    def run(self):
        """Execute the full pipeline."""
        start = datetime.now()
        raw = self.extract("SELECT * FROM orders WHERE processed = false")
        transformed = self.transform(raw)
        self.load(transformed, "analytics.orders_fact")
        print(f"Pipeline completed in {datetime.now() - start}")

Pandas:  45 seconds, 8GB RAM
Polars:  3 seconds, 1.2GB RAM
DuckDB:  2 seconds, 800MB RAM
Spark:   12 seconds, 4GB RAM (overhead for small data)