def unpack_kwargs(kwargs):
 dsk = {}
 kw_pairs = []
 for (key, val) in iteritems(kwargs):
  if isinstance(val, Item):
   dsk.update(val.dask)
   val = val.key
  elif isinstance(val, Base):
   raise NotImplementedError(('%s not supported as kwarg value to Bag.map_partitions' % type(val).__name__))
  kw_pairs.append([key, val])
 return (dsk, kw_pairs)

def unpack_kwargs ( kwargs ) : dsk = { } kw_pairs = [ ] for ( key , val ) in iteritems ( kwargs ) : if isinstance ( val , Item ) : dsk . update ( val . dask ) val = val . key elif isinstance ( val , Base ) : raise NotImplementedError ( ( '%s not supported as kwarg value to Bag.map_partitions' % type ( val ) . __name__ ) ) kw_pairs . append ( [ key , val ] ) return ( dsk , kw_pairs )
extracts dask values from kwargs currently only dask .