def cull(dsk, keys):
    if not isinstance(keys, (list, set)):
        keys = [keys]
    nxt = set(flatten(keys))
    seen = nxt
    while nxt:
        cur = nxt
        nxt = set()
        for item in cur:
            for dep in get_dependencies(dsk, item):
                if dep not in seen:
                    nxt.add(dep)
        seen.update(nxt)
    return dict((k, v) for k, v in dsk.items() if k in seen)
Return new dask with only the tasks required to calculate keys.
dataset/ETHPy150Open dask/dask/dask/optimize.py/cull