Dask read_sql_table ошибки при использовании выражения SQLAlchemy - PullRequest
Новая
       19

Dask read_sql_table ошибки при использовании выражения SQLAlchemy

0 голосов
/ 22 февраля 2019

Я пытаюсь использовать выражение SQLAlchemy с таблицей read_sql_table от dask, чтобы получить набор данных, созданный путем объединения и фильтрации нескольких разных таблиц.Документация указывает, что это должно быть возможно.

(В приведенном ниже примере не включены никакие объединения, поскольку они не нужны для репликации проблемы.)

Я строюв моей строке подключения создайте движок SQLAlchemy и таблицу, соответствующую таблице в моей базе данных.(Я использую PostgreSQL.) 

import dask.dataframe as dd
import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy import Column, MetaData, Table
from sqlalchemy.sql import  select


username = 'username'
password = 'password'
server = 'prod'
database = 'my_db'

connection_string = f'postgresql+psycopg2://{username}:{password}@{server}/{database}'

engine = create_engine(connection_string)

metadata = MetaData()

t = Table('my_table', metadata,
    Column('id'),
    schema='my_schema')

Я могу создать выбор и использовать его с SQLAlchemy без проблем 

>>> s = select([t]).limit(5)
>>> rp = engine.execute(s)
>>> rp.fetchall()

[(3140757,), (3118225,), (3156070,), (3193075,), (3114614,)]

Я также могу кормитьSQLAlchey select для read_sql от panda, который отлично работает 

>>> pd.read_sql(s, connection_string)

id
0   3140757
1   3118225
2   3156070
3   3193075
4   3114614

Однако, когда я передаю тот же select в dask, я получаю ошибку ProgrammingError.Это показывает, что dask оборачивается и вызывает pandas.read_sql, так что вы можете подумать, что он должен работать, но что-то явно не так. 

>>> dd.read_sql_table(s, connection_string, index_col='id')

---------------------------------------------------------------------------
ProgrammingError                          Traceback (most recent call last)
C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\engine\base.py in _execute_context(self, dialect, constructor, statement, parameters, *args)
   1192                         parameters,
-> 1193                         context)
   1194         except BaseException as e:

C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\engine\default.py in do_execute(self, cursor, statement, parameters, context)
    508     def do_execute(self, cursor, statement, parameters, context=None):
--> 509         cursor.execute(statement, parameters)
    510 

ProgrammingError: subquery in FROM must have an alias
LINE 2: FROM (SELECT my_schema.my_table.id AS id 
             ^
HINT:  For example, FROM (SELECT ...) [AS] foo.


The above exception was the direct cause of the following exception:

ProgrammingError                          Traceback (most recent call last)
<ipython-input-5-0db95e60f442> in <module>
----> 1 dd.read_sql_table(s, connection_string, index_col='id')

C:\miniconda3\envs\my_env\lib\site-packages\dask\dataframe\io\sql.py in read_sql_table(table, uri, index_col, divisions, npartitions, limits, columns, bytes_per_chunk, head_rows, schema, meta, engine_kwargs, **kwargs)
    116         # derrive metadata from first few rows
    117         q = sql.select(columns).limit(head_rows).select_from(table)
--> 118         head = pd.read_sql(q, engine, **kwargs)
    119 
    120         if head.empty:

C:\miniconda3\envs\my_env\lib\site-packages\pandas\io\sql.py in read_sql(sql, con, index_col, coerce_float, params, parse_dates, columns, chunksize)
    395             sql, index_col=index_col, params=params,
    396             coerce_float=coerce_float, parse_dates=parse_dates,
--> 397             chunksize=chunksize)
    398 
    399 

C:\miniconda3\envs\my_env\lib\site-packages\pandas\io\sql.py in read_query(self, sql, index_col, coerce_float, parse_dates, params, chunksize)
   1061         args = _convert_params(sql, params)
   1062 
-> 1063         result = self.execute(*args)
   1064         columns = result.keys()
   1065 

C:\miniconda3\envs\my_env\lib\site-packages\pandas\io\sql.py in execute(self, *args, **kwargs)
    952     def execute(self, *args, **kwargs):
    953         """Simple passthrough to SQLAlchemy connectable"""
--> 954         return self.connectable.execute(*args, **kwargs)
    955 
    956     def read_table(self, table_name, index_col=None, coerce_float=True,

C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\engine\base.py in execute(self, statement, *multiparams, **params)
   2073 
   2074         connection = self.contextual_connect(close_with_result=True)
-> 2075         return connection.execute(statement, *multiparams, **params)
   2076 
   2077     def scalar(self, statement, *multiparams, **params):

C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\engine\base.py in execute(self, object, *multiparams, **params)
    946             raise exc.ObjectNotExecutableError(object)
    947         else:
--> 948             return meth(self, multiparams, params)
    949 
    950     def _execute_function(self, func, multiparams, params):

C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\sql\elements.py in _execute_on_connection(self, connection, multiparams, params)
    267     def _execute_on_connection(self, connection, multiparams, params):
    268         if self.supports_execution:
--> 269             return connection._execute_clauseelement(self, multiparams, params)
    270         else:
    271             raise exc.ObjectNotExecutableError(self)

C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\engine\base.py in _execute_clauseelement(self, elem, multiparams, params)
   1058             compiled_sql,
   1059             distilled_params,
-> 1060             compiled_sql, distilled_params
   1061         )
   1062         if self._has_events or self.engine._has_events:

C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\engine\base.py in _execute_context(self, dialect, constructor, statement, parameters, *args)
   1198                 parameters,
   1199                 cursor,
-> 1200                 context)
   1201 
   1202         if self._has_events or self.engine._has_events:

C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\engine\base.py in _handle_dbapi_exception(self, e, statement, parameters, cursor, context)
   1411                 util.raise_from_cause(
   1412                     sqlalchemy_exception,
-> 1413                     exc_info
   1414                 )
   1415             else:

C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\util\compat.py in raise_from_cause(exception, exc_info)
    263     exc_type, exc_value, exc_tb = exc_info
    264     cause = exc_value if exc_value is not exception else None
--> 265     reraise(type(exception), exception, tb=exc_tb, cause=cause)
    266 
    267 if py3k:

C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\util\compat.py in reraise(tp, value, tb, cause)
    246             value.__cause__ = cause
    247         if value.__traceback__ is not tb:
--> 248             raise value.with_traceback(tb)
    249         raise value
    250 

C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\engine\base.py in _execute_context(self, dialect, constructor, statement, parameters, *args)
   1191                         statement,
   1192                         parameters,
-> 1193                         context)
   1194         except BaseException as e:
   1195             self._handle_dbapi_exception(

C:\miniconda3\envs\my_env\lib\site-packages\sqlalchemy\engine\default.py in do_execute(self, cursor, statement, parameters, context)
    507 
    508     def do_execute(self, cursor, statement, parameters, context=None):
--> 509         cursor.execute(statement, parameters)
    510 
    511     def do_execute_no_params(self, cursor, statement, context=None):

ProgrammingError: (psycopg2.ProgrammingError) subquery in FROM must have an alias
LINE 2: FROM (SELECT my_schema.my_table.id AS id 
             ^
HINT:  For example, FROM (SELECT ...) [AS] foo.
 [SQL: 'SELECT id \nFROM (SELECT my_schema.my_table.id AS id \nFROM my_schema.my_table \n LIMIT %(param_1)s) \n LIMIT %(param_2)s'] [parameters: {'param_1': 5, 'param_2': 5}] (Background on this error at: http://sqlalche.me/e/f405)

Ответы [ 2 ]

0 голосов
/ 27 февраля 2019

Для всех остальных, кто сталкивался с этим вопросом.read_sql_table, похоже, не поддерживает этот вариант использования (в настоящее время).Если вы передаете объект SQLAlchemy Select, он в конечном итоге попадает в другой SQLAlchemy Select без псевдонима, что является плохим SQL (по крайней мере, для PostgreSQL).

Если посмотреть на read_sql_table из источника dask, таблица будетобъект Select, который передается в read_sql_table и, как видно, он оборачивается в другой выбор. 

q = sql.select(columns).where(sql.and_(index >= lower, cond)
                              ).select_from(table)

Хорошая новость заключается в том, что функция read_sql_table относительно прямолинейна, а магия - это всего лишь пара строк, которые создаюткадр данных из отложенных объектов.Вам просто нужно написать собственную логику, чтобы разбить запрос на куски 

parts = []
for query_chunk in queries:
    parts.append(delayed(_read_sql_chunk)(q, uri, meta, **kwargs))

return from_delayed(parts, meta, divisions=divisions)


def _read_sql_chunk(q, uri, meta, **kwargs):
    df = pd.read_sql(q, uri, **kwargs)
    if df.empty:
        return meta
    else:
        return df.astype(meta.dtypes.to_dict(), copy=False)
0 голосов
/ 22 февраля 2019

Запрос, отправленный в этой строке, автоматически генерируется SQLAlchemy, поэтому синтаксис должен быть правильным.Однако я заметил, что ваш исходный запрос включает модификатор .limit().Цель строки head = - получить первые несколько строк, чтобы вывести типы.Если в исходном запросе уже есть предложение limit, я вижу, что они могут конфликтовать.Пожалуйста, попробуйте использовать запрос без .limit().

...