When pandas isn’t enough — a real-world story
I often use Python as an automation tool for most daily tasks, from deploy scripts to monitoring alerts. One time I was processing server logs accumulated from a production system — a CSV file nearly 40GB — after running pd.read_csv() my machine froze completely. 16GB of RAM vanished in seconds. That’s when I started seriously looking into Dask.
Dask is a parallel computing library for Python, designed to process data larger than RAM by splitting data into chunks and computing lazily (only executing when results are actually needed). The best part: Dask’s API is almost identical to pandas and NumPy, so you don’t have to learn from scratch.
When should you use Dask?
Before installing, know that Dask is not a silver bullet. It’s a good fit when:
- Dataset is larger than RAM (or takes up more than 50% of RAM with regular pandas)
- You need to leverage multi-core CPU for parallel processing
- Your current workflow uses pandas/NumPy and you want to scale up with minimal refactoring
- Processing many small files in batches (log files, sensor data…)
On the flip side, if the dataset fits comfortably in RAM, regular pandas is still faster because Dask has overhead from task scheduling. Don’t over-engineer.
Installing Dask
Basic installation:
# Install core Dask
pip install dask
# Install complete with distributed scheduler and dashboard
pip install "dask[complete]"
# Or just what you need
pip install "dask[dataframe]" # dask.dataframe
pip install "dask[array]" # dask.array
pip install "dask[distributed]" # distributed scheduler + dashboardVerify installation:
import dask
print(dask.__version__) # 2024.x.xConfiguration and detailed usage
Dask DataFrame — replacing pandas for large files
This is the feature I use most. The API is so close to pandas that you can copy-paste most of your existing code:
import dask.dataframe as dd
# Read large CSV file — Dask splits it into multiple partitions
df = dd.read_csv('server_logs_*.csv') # glob pattern to read multiple files
# Familiar pandas operations
print(df.dtypes) # View immediately — metadata requires no data load
print(df.columns.tolist()) # View column names
# Filter and transform — LAZY: nothing is computed yet
filtered = df[df['status_code'] >= 500]
grouped = filtered.groupby('endpoint')['response_time'].mean()
# .compute() — THIS is where actual computation happens
result = grouped.compute()
print(result)Important tip: only call .compute() once at the end of the pipeline, don’t call it multiple times in the middle — each call means reading the data all over again.
# ❌ Bad — reads data twice
count = df.shape[0].compute()
avg = df['response_time'].mean().compute()
# ✅ Good — use dask.compute() to compute simultaneously
import dask
count, avg = dask.compute(df.shape[0], df['response_time'].mean())Controlling the number of partitions
A partition is Dask’s unit of parallel processing. Too few partitions → CPU underutilization. Too many → task scheduling overhead spikes.
# View current partition count
print(df.npartitions) # Dask usually auto-splits based on file size
# Repartition when needed — e.g., before a group by to balance load
df_balanced = df.repartition(npartitions=20)
# Rule of thumb: partition size ~100MB-1GB is reasonable
# Too small (< 10MB): too much overhead
# Too large (> 2GB): won't fit in worker memoryDask Array — NumPy for large data
If you’re doing machine learning or processing large matrices:
import dask.array as da
import numpy as np
# Create a large array — actually a lazy array with no memory allocated yet
x = da.random.random((100_000, 1000), chunks=(1000, 1000))
# Standard NumPy operations
result = (x ** 2).mean(axis=0)
# Execute
output = result.compute()
print(output.shape) # (1000,)
# Read from .npy or HDF5 file
arr = da.from_array(np.load('large_matrix.npy', mmap_mode='r'), chunks=10000)Configuring the Distributed Scheduler
Dask’s default scheduler uses local threads/processes. For a monitoring dashboard and better control, use distributed:
from dask.distributed import Client
# Start a local cluster — auto-detects CPU core count
client = Client()
print(client) # Prints cluster info + dashboard URL
# Customize number of workers and memory
client = Client(
n_workers=4, # Number of worker processes
threads_per_worker=2, # Threads per worker
memory_limit='4GB' # Max RAM per worker
)
# View cluster info
print(client.scheduler_info())A real-world tip: when running data pipelines overnight on a server, I usually set memory_limit a bit lower than the actual RAM to prevent the OOM killer from stepping in midway.
Reading files efficiently by format
import dask.dataframe as dd
# CSV — most common
df = dd.read_csv('data/*.csv', dtype={'id': 'int32', 'value': 'float32'})
# Parquet — best format for big data (columnar, compressed)
df = dd.read_parquet('data/logs.parquet')
# Convert CSV to Parquet for faster reads next time
df.to_parquet('data/logs_parquet/', write_index=False)
# JSON Lines
df = dd.read_json('events/*.jsonl', lines=True)If you regularly work with large datasets, convert to Parquet early. I once cut load time from 4 minutes down to 20 seconds just with this conversion step.
Performance testing and Monitoring
Dask Dashboard — a tool you can’t skip
After starting Client(), Dask automatically opens a dashboard at http://localhost:8787. The dashboard shows:
- Task Stream: Each running task on a timeline, color-coded by task type
- Progress: Completion percentage of the current computation
- Memory: RAM usage per worker in real time
- Workers: Status, CPU, and memory of each worker
from dask.distributed import Client
client = Client()
# Print dashboard URL to console
print(f"Dashboard: {client.dashboard_link}")
# Or open directly from Python (if using Jupyter)
client # Displays a widget with a clickable linkProfiling and debugging
import dask
# Visualize task graph — see what Dask will compute before running
# Requires graphviz: pip install graphviz
df_filtered = df[df['status'] == 'error'].groupby('host').count()
df_filtered.visualize(filename='task_graph.png')
# Use with performance_report to export an HTML report
from dask.distributed import performance_report
with performance_report(filename='dask_report.html'):
result = df.groupby('endpoint')['latency'].agg(['mean', 'max']).compute()
# View after the run completes
print("Report saved to dask_report.html")Handling memory errors and spill to disk
When a worker runs out of memory, Dask can spill data to disk instead of crashing:
from dask.distributed import Client
client = Client(
n_workers=2,
memory_limit='8GB',
# Spill to disk when memory exceeds threshold
# Configured in dask config
)
# Or set global config
import dask
dask.config.set({
'distributed.worker.memory.target': 0.6, # Start spilling when > 60% memory
'distributed.worker.memory.spill': 0.7, # Spill aggressively when > 70%
'distributed.worker.memory.pause': 0.8, # Pause accepting new tasks when > 80%
'distributed.worker.memory.terminate': 0.95 # Kill worker when > 95%
})Common anti-patterns to avoid
# ❌ Avoid apply() with complex lambdas — slow due to serialize/deserialize
df['new_col'] = df['text'].apply(lambda x: complex_processing(x), meta=('new_col', 'str'))
# ✅ Use map_partitions with vectorized operations
def process_partition(partition):
partition['new_col'] = partition['text'].str.lower().str.strip()
return partition
df = df.map_partitions(process_partition)
# ❌ Avoid iterrows() — can't be parallelized
for idx, row in df.iterrows(): # Extremely slow with Dask
pass
# ✅ Use vectorized operations
df['flagged'] = df['latency'] > 1000Combining Dask with real-world workflows
A complete pipeline I often use to process system logs:
from dask.distributed import Client
import dask.dataframe as dd
def process_logs(log_dir: str, output_path: str):
client = Client(n_workers=4, memory_limit='4GB')
print(f"Dashboard: {client.dashboard_link}")
# Read all log files
df = dd.read_csv(
f'{log_dir}/*.csv',
dtype={'status_code': 'int16', 'latency_ms': 'float32'},
parse_dates=['timestamp']
)
# Processing pipeline
result = (
df
.assign(is_error=df['status_code'] >= 500)
.assign(hour=df['timestamp'].dt.hour)
.groupby(['hour', 'endpoint'])
.agg({'latency_ms': 'mean', 'is_error': 'sum', 'status_code': 'count'})
.rename(columns={'status_code': 'total_requests'})
)
# Save results
result.to_parquet(output_path)
print("Done!")
client.close()
process_logs('/var/log/nginx', './output/daily_stats')Running this script on 40GB of logs, a 16GB RAM machine processed it in about 8 minutes — instead of crashing immediately like plain pandas. The dashboard showed 4 CPU cores running near 100% throughout, with no workers hitting OOM.
Dask isn’t magic — it still needs resources, still has overhead. But for data problems that exceed RAM, it’s the most pragmatic tool available in the Python ecosystem today, especially when a team is already comfortable with pandas.
