Data Manipulation at Scale

import pandas as pd

# Load from various sources
df = pd.read_csv('sales_data.csv')
df = pd.read_parquet('sales_data.parquet')  # Much faster!
df = pd.read_sql('SELECT * FROM sales', connection)

# Basic info
print(df.shape)        # (1000000, 10) - 1M rows, 10 columns
print(df.dtypes)       # Column data types
print(df.info())       # Memory usage, non-null counts
print(df.describe())   # Statistical summary

# Quick preview
print(df.head())       # First 5 rows
print(df.tail(10))     # Last 10 rows
print(df.sample(5))    # 5 random rows

# Column selection
df['customer_id']              # Single column (Series)
df[['customer_id', 'price']]   # Multiple columns (DataFrame)

# Row selection
df.loc[0]                      # By label (first row)
df.iloc[0]                     # By position (first row)
df.loc[0:5]                    # Slice by label (inclusive)
df.iloc[0:5]                   # Slice by position (exclusive)

# Boolean filtering
high_value = df[df['price'] > 100]
recent = df[df['date'] > '2024-01-01']

# Multiple conditions
filtered = df[
    (df['price'] > 50) &
    (df['quantity'] > 1) &
    (df['customer_id'].isin([1001, 1002, 1003]))
]

# Query method (more readable for complex filters)
filtered = df.query('price > 50 and quantity > 1')

# Simple aggregation
total_sales = df['price'].sum()
avg_quantity = df['quantity'].mean()

# Group by single column
by_customer = df.groupby('customer_id').agg({
    'price': 'sum',
    'quantity': 'mean',
    'product_id': 'count'
})

# Group by multiple columns
by_customer_product = df.groupby(['customer_id', 'product_id']).agg({
    'price': ['sum', 'mean', 'std'],
    'quantity': ['sum', 'count']
})

# Custom aggregation functions
def calculate_revenue(group):
    return (group['price'] * group['quantity']).sum()

revenue_by_customer = df.groupby('customer_id').apply(calculate_revenue)

# Multiple aggregations with rename
summary = df.groupby('customer_id').agg(
    total_revenue=('price', 'sum'),
    avg_order_value=('price', 'mean'),
    order_count=('customer_id', 'count'),
    unique_products=('product_id', 'nunique')
)

# Create new columns
df['revenue'] = df['price'] * df['quantity']
df['discount_applied'] = df['price'] < 30

# Apply functions
df['price_category'] = df['price'].apply(
    lambda x: 'High' if x > 100 else 'Medium' if x > 50 else 'Low'
)

# Vectorized operations (much faster!)
df['total_cost'] = df['price'] * df['quantity'] * 1.1  # Add 10% tax

# Map values
category_map = {501: 'Electronics', 502: 'Clothing', 503: 'Food'}
df['category'] = df['product_id'].map(category_map)

# Merge datasets
customers = pd.read_csv('customers.csv')
merged = df.merge(
    customers,
    on='customer_id',
    how='left'  # left, right, inner, outer
)

# Pivot tables (wide format)
pivot = df.pivot_table(
    values='revenue',
    index='date',
    columns='product_id',
    aggfunc='sum',
    fill_value=0
)

# Melt (long format)
long_format = pivot.reset_index().melt(
    id_vars=['date'],
    var_name='product_id',
    value_name='revenue'
)

import dask.dataframe as dd

# Load data (lazy - doesn't load immediately)
ddf = dd.read_csv('large_sales_*.csv')  # Multiple files
ddf = dd.read_parquet('sales_data.parquet')

# All operations are lazy (no computation yet)
filtered = ddf[ddf['price'] > 50]
grouped = filtered.groupby('customer_id')['price'].sum()
result = grouped / 1000  # Still no computation!

# Trigger computation
final_result = result.compute()  # Now it executes in parallel

# Check task graph
print(result.visualize())  # Shows computation plan

# Partitions (how Dask splits data)
print(ddf.npartitions)  # 20 - data split into 20 chunks
print(ddf.divisions)    # Partition boundaries

# Each partition is a Pandas DataFrame
# Dask processes partitions in parallel across cores

from dask.distributed import Client

# Start local cluster (uses all cores)
client = Client()  # Dashboard at http://localhost:8787

# Load large dataset
ddf = dd.read_parquet('sales_100gb.parquet')

# Complex pipeline (all operations are lazy)
result = (
    ddf
    .query('price > 50 and quantity > 1')
    .groupby('customer_id')
    .agg({
        'price': ['sum', 'mean'],
        'quantity': 'sum',
        'product_id': 'nunique'
    })
    .compute()  # Execute in parallel
)

# Persist in memory for reuse
ddf_filtered = ddf[ddf['price'] > 50].persist()

# Now operations on ddf_filtered are fast (already in memory)
high_value_customers = ddf_filtered.groupby('customer_id')['price'].sum()
product_diversity = ddf_filtered.groupby('customer_id')['product_id'].nunique()

# Both use the same persisted data
result1 = high_value_customers.compute()
result2 = product_diversity.compute()

# Custom functions with map_partitions
def custom_analysis(partition):
    """Runs on each partition (Pandas DataFrame)"""
    # Do whatever you want with Pandas
    return partition[partition['price'] > partition['price'].mean()]

result = ddf.map_partitions(custom_analysis).compute()

# PANDAS VERSION (works for small data)
import pandas as pd

df = pd.read_csv('data.csv')
result = (
    df[df['price'] > 50]
    .groupby('customer_id')['price']
    .mean()
)

# DASK VERSION (works for large data) - minimal changes!
import dask.dataframe as dd

ddf = dd.read_csv('data.csv')  # Changed pd to dd
result = (
    ddf[ddf['price'] > 50]
    .groupby('customer_id')['price']
    .mean()
    .compute()  # Added .compute()
)

# What Dask doesn't support well:
# - Operations requiring full dataset sorting
# - Complex window functions
# - Some merge operations with shuffling
# - Operations that break partitioning

# For these, might need to use .compute() to go back to Pandas
difficult_operation = ddf.compute()  # Convert to Pandas
result = difficult_operation.sort_values('complex_column')

# Then convert back to Dask if needed
ddf_result = dd.from_pandas(result, npartitions=20)

import vaex

# Open dataset (instant, no loading!)
df = vaex.open('sales_1billion_rows.hdf5')
df = vaex.open('sales_data.parquet')
df = vaex.open('data/*.csv')  # Multiple files

# Only show what's needed (still not loading all data)
print(df.head())
print(df.tail())
print(df[1000:1010])  # Read only these rows from disk

# Statistics are computed on-demand
print(df.price.mean())      # Fast! Uses efficient algorithms
print(df.price.std())       # Doesn't load all data
print(df.customer_id.unique())  # Scans data, but efficiently

# Convert to Pandas (loads into memory - be careful!)
small_subset = df[df['price'] > 1000]
pandas_df = small_subset.to_pandas_df()  # Now it's in RAM

import vaex

df = vaex.open('sales_data.hdf5')

# Create virtual columns (no memory used!)
df['revenue'] = df.price * df.quantity
df['is_high_value'] = df.price > 100
df['price_category'] = df.price.apply(
    lambda x: 'High' if x > 100 else 'Medium' if x > 50 else 'Low'
)

# Virtual columns computed on-demand
print(df.revenue.mean())  # Computes revenue on-the-fly
print(df.revenue.sum())

# Filter with virtual columns
high_revenue = df[df.revenue > 1000]

# Export to materialize virtual columns
df.export('sales_with_revenue.hdf5')  # Now revenue is a real column

# Expressions are also virtual
df['complex_calc'] = (df.price * df.quantity * 1.1) + df.tax

# Chain operations (all virtual, no memory)
df['normalized_price'] = (df.price - df.price.mean()) / df.price.std()
df['price_zscore'] = df.normalized_price

import vaex

df = vaex.open('sales_1billion_rows.hdf5')

# Simple aggregations (very fast!)
print(df.price.mean())
print(df.price.sum())
print(df.price.std())
print(df.customer_id.nunique())

# Groupby (optimized for large data)
grouped = df.groupby('customer_id').agg({
    'price': 'mean',
    'quantity': 'sum',
    'product_id': vaex.agg.count()
})

# Multiple aggregations
grouped = df.groupby('customer_id', agg={
    'total_revenue': vaex.agg.sum('price'),
    'avg_price': vaex.agg.mean('price'),
    'order_count': vaex.agg.count(),
    'unique_products': vaex.agg.nunique('product_id')
})

# Binning (histograms) - super fast
counts = df.count(binby='price', limits=[0, 1000], shape=100)

# 2D histograms
counts_2d = df.count(
    binby=[df.price, df.quantity],
    limits=[[0, 1000], [0, 100]],
    shape=[100, 50]
)

import polars as pl

# EAGER MODE (executes immediately, like Pandas)
df = pl.read_csv('sales_data.csv')
result = (
    df
    .filter(pl.col('price') > 50)
    .groupby('customer_id')
    .agg(pl.col('price').sum())
)

# LAZY MODE (builds query plan, executes when needed)
df_lazy = pl.scan_csv('sales_data.csv')  # scan_ for lazy
result_lazy = (
    df_lazy
    .filter(pl.col('price') > 50)
    .groupby('customer_id')
    .agg(pl.col('price').sum())
    # Nothing executed yet!
)

# Optimize and execute
result = result_lazy.collect()  # Now it runs (optimized)

# See the query plan
print(result_lazy.explain())

# Example: Lazy is much faster for complex pipelines
# Polars optimizes the entire chain before executing
df_lazy = (
    pl.scan_parquet('sales_*.parquet')
    .filter(pl.col('date') > '2024-01-01')  # Pushed to file read
    .select(['customer_id', 'price'])        # Only reads these columns
    .filter(pl.col('price') > 50)
    .groupby('customer_id')
    .agg(pl.col('price').mean())
)

result = df_lazy.collect()  # Optimized execution

import polars as pl

df = pl.read_csv('sales_data.csv')

# Expression API - chainable, optimizable
result = df.select([
    pl.col('customer_id'),
    pl.col('price').alias('original_price'),
    (pl.col('price') * pl.col('quantity')).alias('revenue'),
    pl.col('price').mean().over('customer_id').alias('avg_customer_price'),
    pl.when(pl.col('price') > 100)
      .then('High')
      .otherwise('Low')
      .alias('price_category')
])

# Powerful aggregations
summary = df.groupby('customer_id').agg([
    pl.col('price').sum().alias('total_spent'),
    pl.col('price').mean().alias('avg_price'),
    pl.col('product_id').n_unique().alias('unique_products'),
    pl.col('quantity').sum().alias('total_quantity'),
    # Conditional aggregation
    pl.col('price').filter(pl.col('price') > 100).sum().alias('high_value_total')
])

# Window functions (over)
df_with_ranks = df.select([
    '*',  # All existing columns
    pl.col('price').rank().over('customer_id').alias('price_rank'),
    pl.col('price').cumsum().over('customer_id').alias('cumulative_spent')
])

# String operations
df = df.with_columns([
    pl.col('customer_name').str.to_uppercase().alias('name_upper'),
    pl.col('customer_name').str.split(' ').alias('name_parts')
])

import polars as pl
import time

# Automatic parallelization (uses all cores)
df = pl.read_csv('large_data.csv')

start = time.time()
result = (
    df
    .groupby('customer_id')
    .agg([
        pl.col('price').sum(),
        pl.col('quantity').mean(),
        pl.col('product_id').n_unique()
    ])
)
elapsed = time.time() - start

# Streaming for larger-than-memory data
result = (
    pl.scan_csv('huge_file.csv')
    .filter(pl.col('price') > 50)
    .groupby('customer_id')
    .agg(pl.col('price').sum())
    .collect(streaming=True)  # Process in chunks
)

# Predicate pushdown (filter at read time)
df = pl.scan_parquet('sales_*.parquet') \
    .filter(pl.col('date').is_between('2024-01-01', '2024-12-31')) \
    .collect()
# Only reads relevant row groups from Parquet!

# Projection pushdown (read only needed columns)
df = pl.scan_csv('data.csv') \
    .select(['customer_id', 'price']) \
    .collect()
# Only reads 2 columns, not all!

# Parallel CSV parsing
df = pl.read_csv('data.csv', n_threads=16)  # Uses 16 threads

# Memory-efficient joins
result = df1.join(df2, on='customer_id', how='left')
# Optimized hash join, multi-threaded

import pandas as pd

df = pd.read_parquet('sales_50gb.parquet')
filtered = df[df['customer_lifetime_value'] > 10000]
result = filtered.groupby('customer_id').agg({
    'revenue': 'sum',
    'order_count': 'count',
    'avg_order_value': 'mean'
})
result.to_csv('high_value_customers.csv')

import dask.dataframe as dd

ddf = dd.read_parquet('sales_50gb.parquet')
filtered = ddf[ddf['customer_lifetime_value'] > 10000]
result = filtered.groupby('customer_id').agg({
    'revenue': 'sum',
    'order_count': 'count',
    'avg_order_value': 'mean'
}).compute()
result.to_csv('high_value_customers.csv')

import vaex

df = vaex.open('sales_50gb.hdf5')  # Instant!
filtered = df[df.customer_lifetime_value > 10000]
result = filtered.groupby('customer_id', agg={
    'revenue': vaex.agg.sum('revenue'),
    'order_count': vaex.agg.count(),
    'avg_order_value': vaex.agg.mean('order_value')
})
result.export('high_value_customers.csv')

import polars as pl

df = pl.scan_parquet('sales_50gb.parquet')
result = (
    df
    .filter(pl.col('customer_lifetime_value') > 10000)
    .groupby('customer_id')
    .agg([
        pl.col('revenue').sum(),
        pl.count().alias('order_count'),
        pl.col('order_value').mean().alias('avg_order_value')
    ])
    .collect()
)
result.write_csv('high_value_customers.csv')