U /eP @sddlZddlZddlZddlmZddlm Z ddl m Z m Z ddl mZmZddlmZdd d Zdd d Zd dZdddZdddZd eeeedddZeddddZdS)!N)compute)methods) from_delayed from_pandas)delayedtokenize) parse_bytes256 MiBc Ks`ddl} t|ts*tdtt|d|dkr:tdt|t| j| jjj fsftdtt||dks|dkr~td|dkr|dkrtd|r|rtd | dkrin| } | j |f| } t|tr| |n| |j |j} | j | d <|dkr^| |}t j|| f| }t|dkr0t|d d S|jd d d|}|dkr^|jdd}|dkr|dkr| j| jj| | jj| g|}t || }|jd\}}|jd}n|\}}t |j}|dkr8| j| jj| g|}t || dd}tt||t |p6d }|j!dkr|t"#t j$||d||%|d}||d<||d<n6|j!dkrt&j'|||d |d#}ntd(|g}|dd|d d}}t)t*||D]j\}\}}|t|d kr| |kn| |k}|+| j,| |k|}|-t.t/|||fd| i| q| 0t1|||dS)ar Read SQL query into a DataFrame. If neither ``divisions`` or ``npartitions`` is given, the memory footprint of the first few rows will be determined, and partitions of size ~256MB will be used. Parameters ---------- sql : SQLAlchemy Selectable SQL query to be executed. TextClause is not supported con : str Full sqlalchemy URI for the database connection index_col : str Column which becomes the index, and defines the partitioning. Should be a indexed column in the SQL server, and any orderable type. If the type is number or time, then partition boundaries can be inferred from ``npartitions`` or ``bytes_per_chunk``; otherwise must supply explicit ``divisions``. divisions: sequence Values of the index column to split the table by. If given, this will override ``npartitions`` and ``bytes_per_chunk``. The divisions are the value boundaries of the index column used to define the partitions. For example, ``divisions=list('acegikmoqsuwz')`` could be used to partition a string column lexographically into 12 partitions, with the implicit assumption that each partition contains similar numbers of records. npartitions : int Number of partitions, if ``divisions`` is not given. Will split the values of the index column linearly between ``limits``, if given, or the column max/min. The index column must be numeric or time for this to work limits: 2-tuple or None Manually give upper and lower range of values for use with ``npartitions``; if None, first fetches max/min from the DB. Upper limit, if given, is inclusive. bytes_per_chunk : str or int If both ``divisions`` and ``npartitions`` is None, this is the target size of each partition, in bytes head_rows : int How many rows to load for inferring the data-types, and memory per row meta : empty DataFrame or None If provided, do not attempt to infer dtypes, but use these, coercing all chunks on load engine_kwargs : dict or None Specific db engine parameters for sqlalchemy kwargs : dict Additional parameters to pass to `pd.read_sql()` Returns ------- dask.dataframe See Also -------- read_sql_table : Read SQL database table into a DataFrame. rNz'con' must be of type str, not 8Note: Dask does not support SQLAlchemy connectables herez)Must specify index column to partition onz2'index_col' must be of type str or sa.Column, not z'Must provide 'meta' if 'head_rows' is 0z=Must provide 'divisions' or 'npartitions' if 'head_rows' is 0z9Must supply either 'divisions' or 'npartitions', not both index_col) npartitionsT)deepindexZmax_1Zcount_1Mz%iS)startendfreq)iuf)dtypezwProvided index column is of type "{}". If divisions is not provided the index column type must be numeric or datetime. engine_kwargs) divisions)2 sqlalchemy isinstancestr TypeErrortype ValueErrorColumnsqlelementsZ ColumnClause create_enginenamelimitpdread_sqllenrZ memory_usagesumZilocselectfuncmaxmin select_fromZsubquerydtypesZSeriesrcountintroundrkindrtolistZ date_range total_secondsnpZlinspaceformat enumeratezipwhereand_appendr_read_sql_chunkdisposer)r#conr rrlimitsbytes_per_chunk head_rowsmetarkwargssaenginerqheadZ bytes_per_rowZminmaxZmaxiZminirr2partsZlowersZuppersrlowerupperZcondrN9/tmp/pip-unpacked-wheel-dbjnr7gq/dask/dataframe/io/sql.pyread_sql_query sD              "   "rPc  sddlddlm} d| kr2tdt| dd| krPtdt| d}ttsntdtt |dk r|D](}t|j tfsztd tt |qzt|tstd tt |d | dkrin| } j |f| } }ttrj |d || d ntdtt ||rDfdd|DnfddjD}t|trx |j|j n |j|j }|jdd|Dkr||| |}tf||||||||| | d | S)a Read SQL database table into a DataFrame. If neither ``divisions`` or ``npartitions`` is given, the memory footprint of the first few rows will be determined, and partitions of size ~256MB will be used. Parameters ---------- table_name : str Name of SQL table in database. con : str Full sqlalchemy URI for the database connection index_col : str Column which becomes the index, and defines the partitioning. Should be a indexed column in the SQL server, and any orderable type. If the type is number or time, then partition boundaries can be inferred from ``npartitions`` or ``bytes_per_chunk``; otherwise must supply explicit ``divisions``. columns : sequence of str or SqlAlchemy column or None Which columns to select; if None, gets all. Note can be a mix of str and SqlAlchemy columns schema : str or None Pass this to sqlalchemy to select which DB schema to use within the URI connection divisions: sequence Values of the index column to split the table by. If given, this will override ``npartitions`` and ``bytes_per_chunk``. The divisions are the value boundaries of the index column used to define the partitions. For example, ``divisions=list('acegikmoqsuwz')`` could be used to partition a string column lexographically into 12 partitions, with the implicit assumption that each partition contains similar numbers of records. npartitions : int Number of partitions, if ``divisions`` is not given. Will split the values of the index column linearly between ``limits``, if given, or the column max/min. The index column must be numeric or time for this to work limits: 2-tuple or None Manually give upper and lower range of values for use with ``npartitions``; if None, first fetches max/min from the DB. Upper limit, if given, is inclusive. bytes_per_chunk : str or int If both ``divisions`` and ``npartitions`` is None, this is the target size of each partition, in bytes head_rows : int How many rows to load for inferring the data-types, and memory per row meta : empty DataFrame or None If provided, do not attempt to infer dtypes, but use these, coercing all chunks on load engine_kwargs : dict or None Specific db engine parameters for sqlalchemy kwargs : dict Additional parameters to pass to `pd.read_sql()` Returns ------- dask.dataframe See Also -------- read_sql_query : Read SQL query into a DataFrame. Examples -------- >>> df = dd.read_sql_table('accounts', 'sqlite:///path/to/bank.db', ... npartitions=10, index_col='id') # doctest: +SKIP rN)r#tablezWThe `table` keyword has been replaced by `table_name`. Please use `table_name` instead.urizGThe `uri` keyword has been replaced by `con`. Please use `con` instead.z&`table_name` must be of type str, not z8`columns` must be of type List[str], and cannot contain z`con` must be of type str, not r T)ZautoloadZ autoload_withschemacs:g|]2}t|tr&|j|jn|j|jqSrN)rrr"columnsr r&.0crG table_namerNrO 7sz"read_sql_table..csg|]}|j|jqSrN)r"r&r rU)rGrNrOrZ@scSsg|] }|jqSrN)r&rUrNrNrOrZHs) r#rAr rrrBrCrDrEr)rr#warningswarnDeprecationWarningpoprrrr r"r%ZMetaDataZTabler@rTr&r>r,r0rP)rYrAr rrrBrTrCrDrSrErrFr#colrHmrqueryrNrXrOread_sql_tablesP            rbcKs.t|trt|||f|St|||f|SdS)a Read SQL query or database table into a DataFrame. This function is a convenience wrapper around ``read_sql_table`` and ``read_sql_query``. It will delegate to the specific function depending on the provided input. A SQL query will be routed to ``read_sql_query``, while a database table name will be routed to ``read_sql_table``. Note that the delegated function might have more specific notes about their functionality not listed here. Parameters ---------- sql : str or SQLAlchemy Selectable Name of SQL table in database or SQL query to be executed. TextClause is not supported con : str Full sqlalchemy URI for the database connection index_col : str Column which becomes the index, and defines the partitioning. Should be a indexed column in the SQL server, and any orderable type. If the type is number or time, then partition boundaries can be inferred from ``npartitions`` or ``bytes_per_chunk``; otherwise must supply explicit ``divisions``. Returns ------- dask.dataframe See Also -------- read_sql_table : Read SQL database table into a DataFrame. read_sql_query : Read SQL query into a DataFrame. N)rrrbrP)r#rAr rFrNrNrOr)\s! r)cKstddl}|pi}|j|f|}tj||f|}|t|dkrF|St|jdkr\|S|j|jddSdS)NrF)copy) rr%r(r)r@r*r1Zto_dictZastype)rIrRrErrFrGrHdfrNrNrOr?s r?cKs>ddl}|pi}|j|f|}|jfd|i|}||S)NrrA)rr%to_sqlr@)drRrrFrGrHrIrNrNrO _to_sql_chunks rgfailTF)r&rR if_existsrc  st|tstdt|dt||| ||||||| d } tt|jf| t| dd| rvfdd|D}nLg}}|D]:}| t t|fd|id d t |fi|d }qt|}| rt |n|Sd S) aStore Dask Dataframe to a SQL table An empty table is created based on the "meta" DataFrame (and conforming to the caller's "if_exists" preference), and then each block calls pd.DataFrame.to_sql (with `if_exists="append"`). Databases supported by SQLAlchemy [1]_ are supported. Tables can be newly created, appended to, or overwritten. Parameters ---------- name : str Name of SQL table. uri : string Full sqlalchemy URI for the database connection schema : str, optional Specify the schema (if database flavor supports this). If None, use default schema. if_exists : {'fail', 'replace', 'append'}, default 'fail' How to behave if the table already exists. * fail: Raise a ValueError. * replace: Drop the table before inserting new values. * append: Insert new values to the existing table. index : bool, default True Write DataFrame index as a column. Uses `index_label` as the column name in the table. index_label : str or sequence, default None Column label for index column(s). If None is given (default) and `index` is True, then the index names are used. A sequence should be given if the DataFrame uses MultiIndex. chunksize : int, optional Specify the number of rows in each batch to be written at a time. By default, all rows will be written at once. dtype : dict or scalar, optional Specifying the datatype for columns. If a dictionary is used, the keys should be the column names and the values should be the SQLAlchemy types or strings for the sqlite3 legacy mode. If a scalar is provided, it will be applied to all columns. method : {None, 'multi', callable}, optional Controls the SQL insertion clause used: * None : Uses standard SQL ``INSERT`` clause (one per row). * 'multi': Pass multiple values in a single ``INSERT`` clause. * callable with signature ``(pd_table, conn, keys, data_iter)``. Details and a sample callable implementation can be found in the section :ref:`insert method `. compute : bool, default True When true, call dask.compute and perform the load into SQL; otherwise, return a Dask object (or array of per-block objects when parallel=True) parallel : bool, default False When true, have each block append itself to the DB table concurrently. This can result in DB rows being in a different order than the source DataFrame's corresponding rows. When false, load each block into the SQL DB in sequence. engine_kwargs : dict or None Specific db engine parameters for sqlalchemy Raises ------ ValueError When the table already exists and `if_exists` is 'fail' (the default). See Also -------- read_sql : Read a DataFrame from a table. Notes ----- Timezone aware datetime columns will be written as ``Timestamp with timezone`` type with SQLAlchemy if supported by the database. Otherwise, the datetimes will be stored as timezone unaware timestamps local to the original timezone. .. versionadded:: 0.24.0 References ---------- .. [1] https://docs.sqlalchemy.org .. [2] https://www.python.org/dev/peps/pep-0249/ Examples -------- Create a table from scratch with 4 rows. >>> import pandas as pd >>> import dask.dataframe as dd >>> df = pd.DataFrame([ {'i':i, 's':str(i)*2 } for i in range(4) ]) >>> ddf = dd.from_pandas(df, npartitions=2) >>> ddf # doctest: +SKIP Dask DataFrame Structure: i s npartitions=2 0 int64 object 2 ... ... 3 ... ... Dask Name: from_pandas, 2 tasks >>> from dask.utils import tmpfile >>> from sqlalchemy import create_engine >>> with tmpfile() as f: ... db = 'sqlite:///%s' %f ... ddf.to_sql('test', db) ... engine = create_engine(db, echo=False) ... result = engine.execute("SELECT * FROM test").fetchall() >>> result [(0, 0, '00'), (1, 1, '11'), (2, 2, '22'), (3, 3, '33')] z!Expected URI to be a string, got .) r&rRrrSrir index_label chunksizermethodr>)ric s4g|],}tt|fdiddt|fiqS)extras dask_key_name to_sql-%s) _extra_depsrgr)rVrfZ meta_taskZ worker_kwargsrNrOrZ4szto_sql..rnrorprN) rrr!r dictrrgZ_metaZ to_delayedr>rqr dask_compute)rdr&rRrSrirrkrlrrmrparallelrrFresultlastrfrNrrrOresP|        re)rncOs |||S)NrN)r-rnargsrFrNrNrOrqUsrq)NNNr r NN) NNNNr r NNN)N)N) NrhTNNNNTFN)r[Znumpyr8Zpandasr(Z dask.baserrtZdask.dataframerZdask.dataframe.io.iorrZ dask.delayedrrZ dask.utilsrrPrbr)r?rgrboolrerqrNrNrNrOsb    . ''   6