icaoberg / Numerical computing - Intro to Dask

Created Sun, 17 May 2026 00:00:00 +0000 Modified Sun, 17 May 2026 17:14:22 -0400

Jupyter Notebook python

Numerical computing - Intro to Dask¶

Dask is a flexible parallel computing library for Python that scales from a laptop to a cluster. It mirrors the APIs of NumPy, Pandas, and scikit-learn so existing code needs minimal changes.

This notebook covers:

What is Dask and how it works
When Dask is the right tool (and when it is not)
Core data structures: dask.array, dask.dataframe, dask.bag
The task graph and lazy evaluation
The Dask scheduler and distributed client
Common operations on Dask DataFrames
Performance tips

1. How Dask Works¶

Dask breaks large computations into a task graph — a directed acyclic graph (DAG) where each node is a small unit of work and edges represent data dependencies.

Large dataset
    └── Partition 0  ─┐
    └── Partition 1  ─┤ → combine → result
    └── Partition 2  ─┘

Key properties:

Lazy evaluation — nothing runs until you call .compute()
Blocked / partitioned — data is split into chunks that fit in RAM
Parallel — partitions are processed concurrently across threads, processes, or machines
API compatible — dask.dataframe mirrors pandas, dask.array mirrors numpy

2. When to Use Dask — and When Not To¶

Dask adds overhead (task scheduling, serialization, coordination). It pays off only when the work is large enough or parallel enough to outweigh that cost.

2.1 Use Dask when ✅¶

Situation	Why Dask helps
Data larger than RAM	Partitioned processing reads only one chunk at a time
Embarrassingly parallel workloads	Each partition is independent; linear speedup is achievable
Multi-core machines / clusters	Distributes work across all available cores or nodes
Iterative file processing (many CSVs, Parquets)	Reads and processes files in parallel
Large-scale ML pipelines (with `dask-ml`)	Scales scikit-learn estimators to big datasets
Streaming / incremental computation	Processes data in chunks without loading everything
You already use pandas/numpy	Drop-in API means low migration cost

2.2 Do NOT use Dask when ❌¶

Situation	Why Dask hurts
Data fits comfortably in RAM	Pandas is faster with zero scheduling overhead
Highly sequential algorithms	Cannot be parallelized; partitioning just adds cost
Many small tasks	Scheduler overhead exceeds compute time per task
Complex joins across all partitions	Requires a shuffle — expensive network/disk I/O
Interactive / exploratory work on small data	`.compute()` latency breaks the REPL flow
Low-latency single queries	Use a database or DuckDB instead
GPU-heavy deep learning	Use PyTorch/TensorFlow distributed, not Dask

Rule of thumb: if pandas finishes in under a few seconds, don't reach for Dask.

2.3 Dask vs. alternatives¶

Tool	Best for
Pandas	Data that fits in RAM, complex transformations, interactive EDA
Dask	Out-of-core / multi-core pandas/numpy on a single machine or cluster
Spark (PySpark)	Very large clusters, SQL-heavy pipelines, enterprise ecosystems
DuckDB	Fast analytical SQL on files (CSV/Parquet) without a cluster
Ray	General distributed Python, ML serving, reinforcement learning
Polars	Fast single-node DataFrame operations with lazy evaluation

3. Installation¶

In [1]:

!pip install dask[complete] -q # includes distributed, dataframe, array, bag

4. Import¶

In [2]:

import dask
import dask.dataframe as dd
import dask.array as da
import dask.bag as db
import pandas as pd
import numpy as np

print(f'dask version: {dask.__version__}')

dask version: 2025.2.0

5. The Task Graph and Lazy Evaluation¶

Dask builds a task graph when you write expressions. Nothing is executed until .compute() is called. This allows Dask to optimize the full pipeline before running it.

In [3]:

# Build a lazy computation — no work happens yet
x = da.from_array(np.arange(1_000_000), chunks=100_000)
result = (x ** 2).mean()

print(type(result))   # dask.array.Array — still lazy
print(result)         # shows the graph description, not the value

<class 'dask.array.core.Array'>
dask.array<mean_agg-aggregate, shape=(), dtype=float64, chunksize=(), chunktype=numpy.ndarray>

In [4]:

# Trigger execution
print(result.compute())   # now the work runs

333332833333.50006

In [5]:

# Visualize the task graph (requires graphviz)
# result.visualize(filename='task_graph.png')

6. Dask DataFrame¶

A Dask DataFrame is a partitioned pandas DataFrame. Each partition is an ordinary pandas DataFrame. Operations on a Dask DataFrame are lazy and produce another Dask DataFrame until .compute() is called.

6.1 Create from a pandas DataFrame¶

In [6]:

pdf = pd.DataFrame({
    'name':   ['Alice', 'Bob', 'Carol', 'Dave', 'Eve',
               'Frank', 'Grace', 'Hank', 'Iris', 'Jack'],
    'dept':   ['Eng', 'HR', 'Eng', 'Finance', 'HR',
               'Eng', 'Finance', 'HR', 'Eng', 'Finance'],
    'salary': [95000, 72000, 105000, 88000, 68000,
               115000, 92000, 78000, 99000, 84000],
    'years':  [5, 3, 8, 6, 2, 10, 4, 7, 9, 1],
})

ddf = dd.from_pandas(pdf, npartitions=2)
print(f'Partitions: {ddf.npartitions}')
ddf

Partitions: 2

Out[6]:

Dask DataFrame Structure:

	name	dept	salary	years
npartitions=2
0	string	string	int64	int64
5	...	...	...	...
9	...	...	...	...

Dask Name: frompandas, 1 expression

6.2 Read from files (the primary use case)¶

In [7]:

# Single CSV
# ddf = dd.read_csv('data.csv')

# Multiple CSVs with a glob pattern — each file becomes a partition
# ddf = dd.read_csv('data/chunk_*.csv')

# Parquet directory (columnar; much faster for large datasets)
# ddf = dd.read_parquet('data/output.parquet')

# Useful parameters (same as pandas):
# dd.read_csv('data/*.csv',
#     blocksize='64MB',    # partition size hint
#     usecols=['a', 'b'],  # read only needed columns
#     dtype={'col': str},
# )

6.3 Explore the DataFrame¶

In [8]:

print(ddf.dtypes)        # instant — metadata only
print(ddf.columns.tolist())
print(ddf.npartitions)

name      string[pyarrow]
dept      string[pyarrow]
salary              int64
years               int64
dtype: object
['name', 'dept', 'salary', 'years']
2

In [9]:

ddf.head()    # returns first 5 rows as a pandas DataFrame (triggers partial compute)

Out[9]:

	name	dept	salary	years
0	Alice	Eng	95000	5
1	Bob	HR	72000	3
2	Carol	Eng	105000	8
3	Dave	Finance	88000	6
4	Eve	HR	68000	2

In [10]:

# .compute() materializes the entire result as a pandas DataFrame
ddf.compute()

Out[10]:

	name	dept	salary	years
0	Alice	Eng	95000	5
1	Bob	HR	72000	3
2	Carol	Eng	105000	8
3	Dave	Finance	88000	6
4	Eve	HR	68000	2
5	Frank	Eng	115000	10
6	Grace	Finance	92000	4
7	Hank	HR	78000	7
8	Iris	Eng	99000	9
9	Jack	Finance	84000	1

6.4 Common operations¶

In [11]:

# Filter rows
high_earners = ddf[ddf['salary'] > 90000]
high_earners.compute()

Out[11]:

	name	dept	salary	years
0	Alice	Eng	95000	5
2	Carol	Eng	105000	8
5	Frank	Eng	115000	10
6	Grace	Finance	92000	4
8	Iris	Eng	99000	9

In [12]:

# Add a new column
ddf['salary_per_year'] = ddf['salary'] / ddf['years']
ddf.head()

Out[12]:

	name	dept	salary	years	salary_per_year
0	Alice	Eng	95000	5	19000.000000
1	Bob	HR	72000	3	24000.000000
2	Carol	Eng	105000	8	13125.000000
3	Dave	Finance	88000	6	14666.666667
4	Eve	HR	68000	2	34000.000000

In [13]:

# GroupBy + aggregation
dept_avg = ddf.groupby('dept')['salary'].mean()
dept_avg.compute()

Out[13]:

dept
Eng        103500.000000
Finance     88000.000000
HR          72666.666667
Name: salary, dtype: float64

In [14]:

# Value counts
ddf['dept'].value_counts().compute()

Out[14]:

dept
Finance    3
HR         3
Eng        4
Name: count, dtype: int64[pyarrow]

In [15]:

# Sort (expensive — triggers a full shuffle)
ddf.sort_values('salary', ascending=False).compute()

Out[15]:

	name	dept	salary	years	salary_per_year
5	Frank	Eng	115000	10	11500.000000
2	Carol	Eng	105000	8	13125.000000
8	Iris	Eng	99000	9	11000.000000
0	Alice	Eng	95000	5	19000.000000
6	Grace	Finance	92000	4	23000.000000
3	Dave	Finance	88000	6	14666.666667
9	Jack	Finance	84000	1	84000.000000
7	Hank	HR	78000	7	11142.857143
1	Bob	HR	72000	3	24000.000000
4	Eve	HR	68000	2	34000.000000

In [16]:

# Drop duplicates
ddf.drop_duplicates(subset=['name']).compute()

Out[16]:

	name	dept	salary	years	salary_per_year
1	Bob	HR	72000	3	24000.000000
3	Dave	Finance	88000	6	14666.666667
4	Eve	HR	68000	2	34000.000000
7	Hank	HR	78000	7	11142.857143
9	Jack	Finance	84000	1	84000.000000
0	Alice	Eng	95000	5	19000.000000
2	Carol	Eng	105000	8	13125.000000
5	Frank	Eng	115000	10	11500.000000
6	Grace	Finance	92000	4	23000.000000
8	Iris	Eng	99000	9	11000.000000

In [17]:

# Merge / join two Dask DataFrames
pdf2 = pd.DataFrame({'dept': ['Eng', 'HR', 'Finance'], 'location': ['Pittsburgh', 'NYC', 'Chicago']})
ddf2 = dd.from_pandas(pdf2, npartitions=1)

merged = ddf.merge(ddf2, on='dept')
merged.compute()

Out[17]:

	name	dept	salary	years	salary_per_year	location
0	Alice	Eng	95000	5	19000.000000	Pittsburgh
1	Bob	HR	72000	3	24000.000000	NYC
2	Carol	Eng	105000	8	13125.000000	Pittsburgh
3	Dave	Finance	88000	6	14666.666667	Chicago
4	Eve	HR	68000	2	34000.000000	NYC
0	Frank	Eng	115000	10	11500.000000	Pittsburgh
1	Grace	Finance	92000	4	23000.000000	Chicago
2	Hank	HR	78000	7	11142.857143	NYC
3	Iris	Eng	99000	9	11000.000000	Pittsburgh
4	Jack	Finance	84000	1	84000.000000	Chicago

7. Dask Array¶

A Dask Array is a partitioned NumPy ndarray. It supports most NumPy operations lazily.

In [18]:

# Create a large lazy array
x = da.random.random((10_000, 10_000), chunks=(1_000, 1_000))
print(f'Shape: {x.shape}, Chunks: {x.chunks[0][0]}')
print(f'Total size: {x.nbytes / 1e6:.0f} MB')

Shape: (10000, 10000), Chunks: 1000
Total size: 800 MB

In [19]:

# Lazy operations — identical to NumPy
result = (x + x.T).mean(axis=0)
print(type(result))   # still dask.array

# Materialize
arr = result.compute()
print(arr.shape)

<class 'dask.array.core.Array'>
(10000,)

8. Dask Bag¶

Dask Bag processes collections of arbitrary Python objects (like map/filter/reduce). It is ideal for unstructured or semi-structured data such as JSON logs.

In [20]:

records = [
    {'user': 'alice', 'action': 'login'},
    {'user': 'bob',   'action': 'purchase'},
    {'user': 'alice', 'action': 'purchase'},
    {'user': 'carol', 'action': 'login'},
]

bag = db.from_sequence(records, npartitions=2)

purchases = bag.filter(lambda r: r['action'] == 'purchase') \
               .map(lambda r: r['user'])

purchases.compute()

Out[20]:

['bob', 'alice']

9. The Distributed Scheduler¶

By default Dask uses a threaded scheduler (good for numpy/array work) or a synchronous scheduler (great for debugging).

For multi-process or cluster execution, use the distributed package.

In [21]:

# Local cluster — uses all available CPU cores
# from dask.distributed import Client
# client = Client()          # spins up a local cluster
# print(client.dashboard_link)  # open in browser to monitor tasks

# Connect to a remote cluster
# client = Client('scheduler-address:8786')

# Close when done
# client.close()

In [22]:

# Choose scheduler explicitly without a Client
import dask

# Threaded (default for dask.array — GIL-releasing operations)
# result.compute(scheduler='threads')

# Multiprocessing (good for pure-Python / pandas work)
# result.compute(scheduler='processes')

# Synchronous — runs everything serially, easiest to debug
# result.compute(scheduler='synchronous')

10. Performance Tips¶

Tip	Why it matters
Use Parquet instead of CSV	Columnar format; read only needed columns; much faster
Select only needed columns early	Reduces data shuffled across partitions
Filter early	Fewer rows per partition speeds up all downstream ops
Avoid `iterrows()` / `apply()`	These are row-wise Python loops — not parallelizable
Persist hot data	`ddf = client.persist(ddf)` keeps partitions in worker RAM
Tune partition size	Aim for 100 MB–1 GB per partition; too small = scheduler overhead, too large = OOM
Profile with the dashboard	Real-time task stream shows bottlenecks
Call `.compute()` once	Combine multiple lazy results with `dask.compute(a, b, c)` to share work

In [23]:

# Compute multiple results in one pass (avoids re-reading data)
mean_salary, count_by_dept = dask.compute(
    ddf['salary'].mean(),
    ddf.groupby('dept')['salary'].count(),
)
print(f'Mean salary: {mean_salary:,.0f}')
print(count_by_dept)

Mean salary: 89,600
dept
Eng        4
Finance    3
HR         3
Name: salary, dtype: int64

11. Benchmarks: Pandas vs. Dask¶

The cells below time identical operations on both libraries at different dataset sizes. Results will vary by machine, but the pattern is consistent:

Small data — pandas wins (Dask scheduling overhead dominates)
Large data — Dask wins (parallel partitions overcome the fixed overhead)

We use %%time for single-cell wall time and timeit.timeit() for averaged runs.

11.1 Setup: generate synthetic datasets¶

In [24]:

import time
import timeit
import numpy as np
import pandas as pd
import dask.dataframe as dd
import dask.array as da

np.random.seed(42)

def make_pdf(n):
    return pd.DataFrame({
        'a': np.random.randn(n),
        'b': np.random.randn(n),
        'group': np.random.choice(['x', 'y', 'z'], size=n),
    })

SMALL  = 10_000       #  ~10 K rows
MEDIUM = 1_000_000    #   ~1 M rows
LARGE  = 10_000_000   #  ~10 M rows

print('Datasets defined — run the cells below to benchmark.')

Datasets defined — run the cells below to benchmark.

11.2 Helper — `bench()` utility¶

In [25]:

def bench(label, fn, repeats=3):
    """Run fn() `repeats` times and print min/mean wall time."""
    times = []
    for _ in range(repeats):
        t0 = time.perf_counter()
        fn()
        times.append(time.perf_counter() - t0)
    print(f'{label:45s}  min={min(times)*1000:7.1f} ms  mean={sum(times)/len(times)*1000:7.1f} ms')

11.3 GroupBy + mean on small data¶

With only 10 K rows, Dask overhead (building the graph, dispatching partitions) is larger than the actual compute time — pandas wins.

In [26]:

pdf_small = make_pdf(SMALL)
ddf_small = dd.from_pandas(pdf_small, npartitions=4)

bench('pandas  groupby mean  (10 K rows)',
      lambda: pdf_small.groupby('group')['a'].mean())

bench('dask    groupby mean  (10 K rows)',
      lambda: ddf_small.groupby('group')['a'].mean().compute())

pandas  groupby mean  (10 K rows)              min=    0.6 ms  mean=    1.7 ms
dask    groupby mean  (10 K rows)              min=   30.3 ms  mean=   33.4 ms

11.4 GroupBy + mean on large data¶

At 10 M rows the computation is heavy enough that parallel partitions pay off.

In [27]:

pdf_large = make_pdf(LARGE)
ddf_large = dd.from_pandas(pdf_large, npartitions=8)

bench('pandas  groupby mean  (10 M rows)',
      lambda: pdf_large.groupby('group')['a'].mean())

bench('dask    groupby mean  (10 M rows)',
      lambda: ddf_large.groupby('group')['a'].mean().compute())

pandas  groupby mean  (10 M rows)              min=  417.5 ms  mean=  456.3 ms
dask    groupby mean  (10 M rows)              min=  588.2 ms  mean=  648.7 ms

11.5 Column arithmetic on different sizes¶

Element-wise arithmetic is memory-bound; Dask helps only when data exceeds L3 cache.

In [28]:

for n, label in [(SMALL, '10 K'), (MEDIUM, '1 M'), (LARGE, '10 M')]:
    pdf = make_pdf(n)
    ddf = dd.from_pandas(pdf, npartitions=8)
    bench(f'pandas  a*b + a  ({label:>4} rows)',
          lambda p=pdf: p['a'] * p['b'] + p['a'])
    bench(f'dask    a*b + a  ({label:>4} rows)',
          lambda d=ddf: (d['a'] * d['b'] + d['a']).compute())
    print()

pandas  a*b + a  (10 K rows)                   min=    0.1 ms  mean=    0.2 ms
dask    a*b + a  (10 K rows)                   min=   21.2 ms  mean=   22.2 ms

pandas  a*b + a  ( 1 M rows)                   min=    2.4 ms  mean=    2.9 ms
dask    a*b + a  ( 1 M rows)                   min=   63.4 ms  mean=   68.0 ms

pandas  a*b + a  (10 M rows)                   min=   65.4 ms  mean=   74.3 ms
dask    a*b + a  (10 M rows)                   min=  601.1 ms  mean=  614.0 ms

11.6 NumPy array vs. Dask array¶

For pure-NumPy work on a single machine, Dask benefits appear only for arrays that don't fit in RAM or when using multi-core CPUs for GIL-releasing ufuncs.

In [ ]:

sizes = [1_000_000, 10_000_000, 100_000_000]

for n in sizes:
    np_arr = np.random.randn(n)
    da_arr = da.from_array(np_arr, chunks=n // 8)

    bench(f'numpy  sum  (n={n:>11,})',
          lambda a=np_arr: np.sum(a))
    bench(f'dask   sum  (n={n:>11,})',
          lambda a=da_arr: a.sum().compute())
    print()

numpy  sum  (n=  1,000,000)                    min=    0.5 ms  mean=    0.6 ms
dask   sum  (n=  1,000,000)                    min=    5.9 ms  mean=    6.5 ms

numpy  sum  (n= 10,000,000)                    min=    4.4 ms  mean=    5.3 ms
dask   sum  (n= 10,000,000)                    min=    7.3 ms  mean=    7.8 ms

11.7 Reading multiple CSV files¶

Dask reads files in parallel — the more files, the bigger the advantage. We write 8 small CSVs to a temp directory, then read them with both libraries.