TopolearnWhen the pandas API in Dask doesn't cover what you need, map_partitions lets you use any pandas code in parallel.
df.map_partitions(func, *args, **kwargs) calls func on each partition (a pandas DataFrame) and stitches the results back into a Dask DataFrame or Series. Anything you can do to a pandas DataFrame, including calling other libraries, can run in parallel this way.
Dask must know the output's columns and dtypes before running anything, so it builds an empty sample called meta. By default it calls your function on a tiny fake partition to guess. That can fail or guess wrong, for example when the function relies on real values. Pass meta= explicitly:
- for a DataFrame:
meta={"a": "int64", "b": "string"}or an emptypd.DataFramewith the right dtypes - for a Series:
meta=("name", "float64")
Important: the function sees one partition at a time. It must not need rows from other partitions. A per-row cleanup, a vectorized feature or a model's predict are fine. A global rank, a running total across partitions, or a groupby that assumes all of a group is present are not, unless the data is partitioned by that key.
map_partitions is also a good place for setup cost you want once per partition rather than per row, like compiling a regex or loading a small lookup table. Related helpers: df.map_overlap for windows that need neighbouring rows, and df.partitions[i] to select single partitions.
Example
import dask.dataframe as dd
import pandas as pd
pdf = pd.DataFrame({"email": ["Ana@X.com", "ben@y.org", "cy@X.COM", "di@z.net"],
"spend": [10.0, 20.0, 30.0, 40.0]})
df = dd.from_pandas(pdf, npartitions=2)
def enrich(part: pd.DataFrame) -> pd.DataFrame:
domain = part["email"].str.split("@").str[1].str.lower()
return part.assign(domain=domain, vip=part["spend"] > 25)
meta = {"email": "string", "spend": "float64", "domain": "string", "vip": "bool"}
out = df.map_partitions(enrich, meta=meta)
print(out.dtypes.astype(str).to_dict())
# {'email': 'string', 'spend': 'float64', 'domain': 'string', 'vip': 'bool'}
print(out.groupby("domain").spend.sum().compute().sort_index().to_dict())
# {'x.com': 40.0, 'y.org': 20.0, 'z.net': 40.0}Remember
map_partitionsruns any pandas function on each partition in parallel- Pass
meta=so Dask knows the output schema without guessing - The function must not need rows from other partitions
Watch out for
- Computing a global statistic (like a rank or z-score) inside map_partitions, which only sees one partition.