U /e)E@sTddlZddlZddlmZddlmZddlmZddlZddlm Z m Z ddl m Z ddl mZddlmZmZmZmZdd lmZdd lmZdd lmZmZdd lmZdd lmZmZddl m!Z!e"de#Z$d)ddZ%d*ddZ&dZ'dZ(GdddeZ)e*dd+ddZ+d d!Z,d"d#Z-d$d%Z.d&d'Z/dd(lm0Z0e&j1e0j&_1dS),Nfnmatch)glob)warn)build_name_functionstringify_path)merge)config)compute_as_if_collection get_schedulernamed_schedulerstokenize)dataframe_creation_dispatch) DataFrame)_linkfrom_map)DataFrameIOFunction)Delayeddelayed)get_scheduler_lockZ processescCs.|r |z|||W5|r(|XdS)z8A wrapper function around pd_to_hdf that enables lockingN)acquirerelease) pd_to_hdflockargskwargsr9/tmp/pip-unpacked-wheel-dbjnr7gq/dask/dataframe/io/hdf.py _pd_to_hdfs raFTc  s&| dkr i} dtj|jj} d} d} t|}t|trr|d|ddkr\t ddd}d|krd } n|ddkrt d d d}d|krd } d | kr| d d krt d|dkrt ddkrt |j d| r| sfddt |j D}|t |krtd|dkr>tdds>| r>| r>d}t|g|d}|dkr|| sbd}n| sx|tk rxd}nd }|rt||d}| d||dtd}t| ||jdf||||d|g| fdf<| }| rd|d<| rd|d<g}t d|j D]}|}||||qt d|j D]h}|}t| ||j|f||||d|g|f}| r| r|dnd}t|f|f}||f<q>t|j| r| r҈|j dfg}nfddt |j D}|r tt|fd|i| |St fdd|DSdS) a Store Dask Dataframe to Hierarchical Data Format (HDF) files This is a parallel version of the Pandas function of the same name. Please see the Pandas docstring for more detailed information about shared keyword arguments. This function differs from the Pandas version by saving the many partitions of a Dask DataFrame in parallel, either to many files, or to many datasets within the same file. You may specify this parallelism with an asterix ``*`` within the filename or datapath, and an optional ``name_function``. The asterix will be replaced with an increasing sequence of integers starting from ``0`` or with the result of calling ``name_function`` on each of those integers. This function only supports the Pandas ``'table'`` format, not the more specialized ``'fixed'`` format. Parameters ---------- path : string, pathlib.Path Path to a target filename. Supports strings, ``pathlib.Path``, or any object implementing the ``__fspath__`` protocol. May contain a ``*`` to denote many filenames. key : string Datapath within the files. May contain a ``*`` to denote many locations name_function : function A function to convert the ``*`` in the above options to a string. Should take in a number from 0 to the number of partitions and return a string. (see examples below) compute : bool Whether or not to execute immediately. If False then this returns a ``dask.Delayed`` value. lock : bool, Lock, optional Lock to use to prevent concurrency issues. By default a ``threading.Lock``, ``multiprocessing.Lock`` or ``SerializableLock`` will be used depending on your scheduler if a lock is required. See dask.utils.get_scheduler_lock for more information about lock selection. scheduler : string The scheduler to use, like "threads" or "processes" **other: See pandas.to_hdf for more information Examples -------- Save Data to a single file >>> df.to_hdf('output.hdf', '/data') # doctest: +SKIP Save data to multiple datapaths within the same file: >>> df.to_hdf('output.hdf', '/data-*') # doctest: +SKIP Save data to multiple files: >>> df.to_hdf('output-*.hdf', '/data') # doctest: +SKIP Save data to multiple files, using the multiprocessing scheduler: >>> df.to_hdf('output-*.hdf', '/data', scheduler='processes') # doctest: +SKIP Specify custom naming scheme. This writes files as '2000-01-01.hdf', '2000-01-02.hdf', '2000-01-03.hdf', etc.. >>> from datetime import date, timedelta >>> base = date(year=2000, month=1, day=1) >>> def name_function(i): ... ''' Convert integer 0 to n to a string ''' ... return base + timedelta(days=i) >>> df.to_hdf('*.hdf', '/data', name_function=name_function) # doctest: +SKIP Returns ------- filenames : list Returned if ``compute`` is True. List of file names that each partition is saved to. delayed : dask.Delayed Returned if ``compute`` is False. Delayed object to execute ``to_hdf`` when computed. See Also -------- read_hdf: to_parquet: Nzto-hdf-T*zBA maximum of one asterisk is accepted in file path and dataset keycSs |d|S)Nr )replace)pathi_namerrrzto_hdf..Fz4A maximum of one asterisk is accepted in dataset keycSs|SNr)r#_rrrr%r&format)ttablez.Dask only support 'table' format in hdf files.)rwzr+z$Mode must be one of 'a', 'w' or 'r+'csg|] }|qSrr.0i) name_functionrr szto_hdf..zWTo preserve order between partitions name_function must preserve the order of its input schedulerzsingle-threaded) collectionsr2)r2r+)r)modeappendrrr4r5csg|] }|fqSrrr-)namerrr1scsg|]}t|qSr)rr.k)dskrrr1 s)!uuidZuuid1hexZ_partition_typeto_hdfr isinstancestrcount ValueErrorrZ npartitionsrangesortedrr getr MP_GETrupdatedictr_namer"copyr5rrdaskr rr)Zdfr#keyr4r5r2r0ZcomputerZ dask_kwargsrrZ single_fileZ single_nodeZfmt_objZformatted_namesZ _actual_getr$Zkwargs2 filenamesr/ZtaskZlink_depkeysr)r9r6r0rr<*sc        r<z This HDFStore is not partitionable and can only be use monolithically with pandas. In the future when creating HDFStores use the ``format='table'`` option to ensure that your dataset can be parallelizedz The start and stop keywords are not supported when reading from more than one file/dataset. The combination is ambiguous because it could be interpreted as the starting and stopping index per file, or starting and stopping index of the global dataset.c@s4eZdZdZddZeddZddZdd Zd S) HDFFunctionWrapperzj HDF5 Function-Wrapper Class Reads HDF5 data from disk to produce a partition (given a key). cCs8||_||_||_||_|r4|dkr4t|d|i|_dS)Nr!columns)_columnsr common_kwargsdimr)selfrNrQrrPrrr__init__"s  zHDFFunctionWrapper.__init__cCs|jSr')rO)rRrrrrN*szHDFFunctionWrapper.columnscCs"||jkr|St||j|j|jS)zUReturn a new HDFFunctionWrapper object with a sub-column projection. )rNrMrQrrP)rRrNrrrproject_columns.s z"HDFFunctionWrapper.project_columnscCsN|\}}}|jr|jztj||ft|j|}W5|jrH|jX|S)zRead from hdf5 file with a lock)rrrpdread_hdfrrP)rRpartr#rJrresultrrr__call__6s   zHDFFunctionWrapper.__call__N) __name__ __module__ __qualname____doc__rSpropertyrNrTrYrrrrrMs  rMpandas@Brc Cs|dkrt}|dr|nd|}t|}t|trDtt|} n|} t|tsft| dkrftd| rvt| dkrt d|| D]D} zt j | } Wntt fk rd} YnX| st d| q|dks|dk rt| d krtt|dkrtd |dks|dk r&|r&td tj| d|d } t|| d} W5QRXztj| d| |dd }Wn*tk rtj| d| |d }YnX|dk r||}|jd kr|j|d}nd|i}t| ||||||\}}tt||j|||||dt| ||||||ddS)a Read HDF files into a Dask DataFrame Read hdf files into a dask dataframe. This function is like ``pandas.read_hdf``, except it can read from a single large file, or from multiple files, or from multiple keys from the same file. Parameters ---------- pattern : string, pathlib.Path, list File pattern (string), pathlib.Path, buffer to read from, or list of file paths. Can contain wildcards. key : group identifier in the store. Can contain wildcards start : optional, integer (defaults to 0), row number to start at stop : optional, integer (defaults to None, the last row), row number to stop at columns : list of columns, optional A list of columns that if not None, will limit the return columns (default is None) chunksize : positive integer, optional Maximal number of rows per partition (default is 1000000). sorted_index : boolean, optional Option to specify whether or not the input hdf files have a sorted index (default is False). lock : boolean, optional Option to use a lock to prevent concurrency issues (default is True). mode : {'a', 'r', 'r+'}, default 'r'. Mode to use when opening file(s). 'r' Read-only; no data can be modified. 'a' Append; an existing file is opened for reading and writing, and if the file does not exist it is created. 'r+' It is similar to 'a', but the file must already exist. Returns ------- dask.DataFrame Examples -------- Load single file >>> dd.read_hdf('myfile.1.hdf5', '/x') # doctest: +SKIP Load multiple files >>> dd.read_hdf('myfile.*.hdf5', '/x') # doctest: +SKIP >>> dd.read_hdf(['myfile.1.hdf5', 'myfile.2.hdf5'], '/x') # doctest: +SKIP Load multiple datasets >>> dd.read_hdf('myfile.1.hdf5', '/*') # doctest: +SKIP T/rzNo files providedzFile(s) not found: Fz,File not found or insufficient permissions: Nr!z$Chunksize must be a positive integerz_When assuming pre-partitioned data, data must be read in its entirety using the same chunksizesr4)r4stop)r6r4r4zread-hdf)meta divisionslabeltokenZenforce_metadata)r startswithrr=r>rBrlenr@OSErrorosr#exists TypeErrorNotImplementedErrorread_hdf_error_msgrUHDFStore _expand_keyrV IndexErrorndimr6 _build_partsrrMr )patternrJstartrdrN chunksize sorted_indexrr4pathsr#rmhdfZmeta_keyrerPpartsrfrrrrVDsnC     rVc Csg}g}|D]l} t| |||||\} } } t| | | D]B\} }}|rX|rX|dd|}n|r`|}|t| | |||q4q ||pdgt|dfS)zG Build the list of partition inputs and divisions for read_hdf Nr!)_get_keys_stops_divisionszipextend_one_path_one_keyrj)rzrJrwrdrxryr4r|Zglobal_divisionsr#rLstopsrfr8sdrrrrus$ rucs:||krtd||fddtt||DS)zo Get the DataFrame corresponding to one path and one key (which should not contain any wildcards). z?Start row number ({}) is above or equal to stop row number ({})cs$g|]\}}||dfqS)rwrdr)r.r/rrxrJr#rrr1sz%_one_path_one_key..)r@r) enumeraterA)r#rJrwrdrxrrrrsrcsZddl}|sgn<fdd|Dfdd|jjdddDS) Nrcsg|]}t|r|qSrrr7)rJrrr1s z_expand_key..c3s4|],}t|jr|jdkr|jkr|jVqdS)r+N)rZ _v_pathnameZ_v_name)r.nrJrLrr s    z_expand_key..rbZTable) classname)r has_magicrLr_handleZ walk_nodes)rJr{rrrrrrs rrc stj||d}g}g}t||} | D]} || jdkrFtt|dkr\|jn&|jkrxt d jn |||rԇfddt dj|D} j djd jd d} | | || q&|dq&W5QRX| ||fS) z Get the "keys" or group identifiers which match the given key, which can contain wildcards (see _expand_path). This uses the hdf file identified by the given path. Also get the index of the last row of data for each matched key. rcr+Nz0Stop keyword exceeds dataset number of rows ({})cs$g|]}jd||dddqS)indexr!rr) read_column)r.rwZstorerrrr1sz-_get_keys_stops_divisions..rrr!r) rUrqrrZ get_storerZ format_typerndont_use_fixed_error_messager5Znrowsr@r)rAr) r#rJrdryrxr4r{rrfrLr8divisionZ division_endrrrr~s@         r~)_Frame)N)rFNNTNN)rNNr`FTra)2rlr:rrwarningsrr_rUZ fsspec.utilsrrZtlzrrIr Z dask.baser r r r Zdask.dataframe.backendsrZdask.dataframe.corerZdask.dataframe.io.iorrZdask.dataframe.io.utilsrZ dask.delayedrrZ dask.utilsrrCobjectrDrr<rrprMZregister_inplacerVrurrrr~rr]rrrrsZ           d ) )