def to_hdf(df, path, key, mode='a', append=False, get=None, name_function=None, compute=True, lock=None, dask_kwargs={}, **kwargs):
 name = ('to-hdf-' + uuid.uuid1().hex)
 pd_to_hdf = getattr(df._partition_type, 'to_hdf')
 single_file = True
 single_node = True
 if isinstance(path, str):
  if ((path.count('*') + key.count('*')) > 1):
   raise ValueError('A maximum of one asterisk is accepted in file path and dataset key')
  fmt_obj = (lambda path, i_name: path.replace('*', i_name))
  if ('*' in path):
   single_file = False
 else:
  if (key.count('*') > 1):
   raise ValueError('A maximum of one asterisk is accepted in dataset key')
  fmt_obj = (lambda path, _: path)
 if ('*' in key):
  single_node = False
 if (('format' in kwargs) and (kwargs['format'] != 'table')):
  raise ValueError("Dask only support 'table' format in hdf files.")
 if (mode not in ('a', 'w', 'r+')):
  raise ValueError("Mode must be one of 'a', 'w' or 'r+'")
 if (name_function is None):
  name_function = build_name_function((df.npartitions - 1))
 if (not (single_file and single_node)):
  formatted_names = [name_function(i) for i in range(df.npartitions)]
  if (formatted_names != sorted(formatted_names)):
   warn('To preserve order between partitions name_function must preserve the order of its input')
 if ((get is None) and ('get' not in _globals) and single_node and single_file):
  get = get_sync
 _actual_get = effective_get(get, df)
 if (lock is None):
  if (not single_node):
   lock = True
  elif ((not single_file) and (_actual_get is not multiprocessing.get)):
   lock = True
  else:
   lock = False
 if lock:
  lock = get_scheduler_lock(get, df)
 kwargs.update({'format': 'table', 'mode': mode, 'append': append})
 dsk = dict()
 i_name = name_function(0)
 dsk[(name, 0)] = (_pd_to_hdf, pd_to_hdf, lock, [(df._name, 0), fmt_obj(path, i_name), key.replace('*', i_name)], kwargs)
 kwargs2 = kwargs.copy()
 if single_file:
  kwargs2['mode'] = 'a'
 if single_node:
  kwargs2['append'] = True
 for i in range(1, df.npartitions):
  i_name = name_function(i)
  task = (_pd_to_hdf, pd_to_hdf, lock, [(df._name, i), fmt_obj(path, i_name), key.replace('*', i_name)], kwargs2)
  if single_file:
   link_dep = ((i - 1) if single_node else 0)
   task = (_link, (name, link_dep), task)
  dsk[(name, i)] = task
 dsk = merge(df.dask, dsk)
 if (single_file and single_node):
  keys = [(name, (df.npartitions - 1))]
 else:
  keys = [(name, i) for i in range(df.npartitions)]
 if compute:
  return DataFrame._get(dsk, keys, get=get, **dask_kwargs)
 else:
  return delayed([Delayed(k, [dsk]) for k in keys])

def to_hdf ( df , path , key , mode = 'a' , append = False , get = None , name_function = None , compute = True , lock = None , dask_kwargs = { } , ** kwargs ) : name = ( 'to-hdf-' + uuid . uuid1 ( ) . hex ) pd_to_hdf = getattr ( df . _partition_type , 'to_hdf' ) single_file = True single_node = True if isinstance ( path , str ) : if ( ( path . count ( '*' ) + key . count ( '*' ) ) > 1 ) : raise ValueError ( 'A maximum of one asterisk is accepted in file path and dataset key' ) fmt_obj = ( lambda path , i_name : path . replace ( '*' , i_name ) ) if ( '*' in path ) : single_file = False else : if ( key . count ( '*' ) > 1 ) : raise ValueError ( 'A maximum of one asterisk is accepted in dataset key' ) fmt_obj = ( lambda path , _ : path ) if ( '*' in key ) : single_node = False if ( ( 'format' in kwargs ) and ( kwargs [ 'format' ] != 'table' ) ) : raise ValueError ( "Dask only support 'table' format in hdf files." ) if ( mode not in ( 'a' , 'w' , 'r+' ) ) : raise ValueError ( "Mode must be one of 'a', 'w' or 'r+'" ) if ( name_function is None ) : name_function = build_name_function ( ( df . npartitions - 1 ) ) if ( not ( single_file and single_node ) ) : formatted_names = [ name_function ( i ) for i in range ( df . npartitions ) ] if ( formatted_names != sorted ( formatted_names ) ) : warn ( 'To preserve order between partitions name_function must preserve the order of its input' ) if ( ( get is None ) and ( 'get' not in _globals ) and single_node and single_file ) : get = get_sync _actual_get = effective_get ( get , df ) if ( lock is None ) : if ( not single_node ) : lock = True elif ( ( not single_file ) and ( _actual_get is not multiprocessing . get ) ) : lock = True else : lock = False if lock : lock = get_scheduler_lock ( get , df ) kwargs . update ( { 'format' : 'table' , 'mode' : mode , 'append' : append } ) dsk = dict ( ) i_name = name_function ( 0 ) dsk [ ( name , 0 ) ] = ( _pd_to_hdf , pd_to_hdf , lock , [ ( df . _name , 0 ) , fmt_obj ( path , i_name ) , key . replace ( '*' , i_name ) ] , kwargs ) kwargs2 = kwargs . copy ( ) if single_file : kwargs2 [ 'mode' ] = 'a' if single_node : kwargs2 [ 'append' ] = True for i in range ( 1 , df . npartitions ) : i_name = name_function ( i ) task = ( _pd_to_hdf , pd_to_hdf , lock , [ ( df . _name , i ) , fmt_obj ( path , i_name ) , key . replace ( '*' , i_name ) ] , kwargs2 ) if single_file : link_dep = ( ( i - 1 ) if single_node else 0 ) task = ( _link , ( name , link_dep ) , task ) dsk [ ( name , i ) ] = task dsk = merge ( df . dask , dsk ) if ( single_file and single_node ) : keys = [ ( name , ( df . npartitions - 1 ) ) ] else : keys = [ ( name , i ) for i in range ( df . npartitions ) ] if compute : return DataFrame . _get ( dsk , keys , get = get , ** dask_kwargs ) else : return delayed ( [ Delayed ( k , [ dsk ] ) for k in keys ] )
store this object .