def cull(dsk, keys):
 if (not isinstance(keys, (list, set))):
  keys = [keys]
 out_keys = []
 seen = set()
 dependencies = dict()
 work = list(set(flatten(keys)))
 while work:
  new_work = []
  out_keys += work
  deps = [(k, get_dependencies(dsk, k, as_list=True)) for k in work]
  dependencies.update(deps)
  for (_, deplist) in deps:
   for d in deplist:
    if (d not in seen):
     seen.add(d)
     new_work.append(d)
  work = new_work
 out = {k: dsk[k] for k in out_keys}
 return (out, dependencies)

def cull ( dsk , keys ) : if ( not isinstance ( keys , ( list , set ) ) ) : keys = [ keys ] out_keys = [ ] seen = set ( ) dependencies = dict ( ) work = list ( set ( flatten ( keys ) ) ) while work : new_work = [ ] out_keys += work deps = [ ( k , get_dependencies ( dsk , k , as_list = True ) ) for k in work ] dependencies . update ( deps ) for ( _ , deplist ) in deps : for d in deplist : if ( d not in seen ) : seen . add ( d ) new_work . append ( d ) work = new_work out = { k : dsk [ k ] for k in out_keys } return ( out , dependencies )
return new dask with only the tasks required to calculate keys .