Dask

Dask splits work on data into many small pieces and runs them as a graph of tasks. dask.dataframe mirrors the pandas API on a table cut into partitions, dask.array does the same for NumPy arrays cut into chunks, and dask.delayed turns your own functions into tasks. Operations are lazy: they build the graph, and nothing runs until you call .compute(). On a laptop or a cluster, people use it for data too big for memory and to spread work across cores. On this page it runs in your browser, so you can try it without installing anything. Run the example first, then paste any snippet below into a new cell to try it.

What the example does

The example imports pyarrow first because dask.dataframe needs it. It builds 10,000 hourly readings in pandas, with timestamps from pd.date_range() and random temperatures and precipitation. dd.from_pandas(df, npartitions=10) splits the table into 10 partitions of 1,000 rows, each an ordinary pandas DataFrame. .dt.date drops the time of day, and groupby("date").temperature.mean() describes the daily average without computing it. .compute() runs the tasks and returns a pandas Series with one value per day, 417 in all, and the last line shows the first five. The data has no seed, so the numbers change on every run.

Filter and group a DataFrame

A Dask DataFrame takes most pandas methods. Printing a result before compute() shows its structure, with ... where the values will be:

import numpy as np
import pandas as pd
import pyarrow  # dask.dataframe needs pyarrow
import dask.dataframe as dd

rng = np.random.default_rng(seed=0)
df = pd.DataFrame({
    "store": rng.choice(["north", "south", "east"], 3000),
    "amount": rng.integers(5, 200, 3000),
})
ddf = dd.from_pandas(df, npartitions=6)

big = ddf[ddf.amount > 100]
summary = big.groupby("store").amount.agg(["count", "mean"])
print(summary)                     # the plan, no numbers yet
print(summary.compute().round(1))  # a pandas DataFrame

Read many CSV files as one table

dd.read_csv() accepts a wildcard and treats all matching files as one DataFrame. This writes three monthly files first:

import pandas as pd
import pyarrow  # dask.dataframe needs pyarrow
import dask.dataframe as dd

for month in [1, 2, 3]:
    pd.DataFrame({
        "month": month,
        "product": ["tea", "coffee", "cocoa"],
        "units": [10 * month, 20 + month, 5],
    }).to_csv(f"sales-{month}.csv", index=False)

ddf = dd.read_csv("sales-*.csv")
print(ddf.npartitions)
print(ddf.groupby("product").units.sum().compute())

Small files like these become one partition each.

Split an array into chunks

da.from_array() cuts a NumPy array into blocks. Reductions such as mean() work block by block and then combine the partial results:

import numpy as np
import dask.array as da

rng = np.random.default_rng(seed=0)
data = rng.normal(10, 2, size=(4000, 300))
x = da.from_array(data, chunks=(1000, 300))    # 4 blocks of 1,000 rows

col_means = x.mean(axis=0)
print(x.numblocks)
print(col_means)                               # lazy until compute()
result = col_means.compute()
print(result[:3].round(3), np.allclose(result, data.mean(axis=0)))

Turn your own functions into tasks

@dask.delayed makes a function return a placeholder instead of running. Calls on placeholders chain into a graph, and compute() runs it:

import dask

@dask.delayed
def load(n):
    return list(range(n))

@dask.delayed
def total(values):
    return sum(values)

parts = [total(load(n)) for n in [10, 20, 30]]
grand = dask.delayed(sum)(parts)

print(type(grand).__name__)   # Delayed: nothing has run yet
print(grand.compute())        # 45 + 190 + 435
print(dask.compute(*parts))   # several results in one pass

Good to know

• Python in the browser cannot start threads or processes. Dask checks for this platform and registers only its synchronous scheduler, so every task runs one at a time in the main thread, and compute(scheduler="threads") raises a ValueError. Partitions still work, but they are processed in turn, not in parallel.
• pyarrow is not installed along with Dask here. Import it before dask.dataframe, as the example does. Otherwise import dask.dataframe fails with ModuleNotFoundError: No module named 'pyarrow'.
• head() looks only at the first partition. If that holds fewer rows than you ask for, you get fewer rows and a warning. Pass npartitions=-1 to search all of them.