python - 使用 SQLAlchemy 表达式时出现 Dask read_sql_table 错误
问题描述
我正在尝试将 SQLAlchemy 表达式与 dask 的 read_sql_table 一起使用,以降低通过连接和过滤几个不同表创建的数据集。文档表明这应该是可能的。
(下面的示例不包括任何连接,因为它们不需要复制问题。)
我建立我的连接字符串,创建一个 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 选择提供给 panda 的 read_sql,效果很好
>>> pd.read_sql(s, connection_string)
id
0 3140757
1 3118225
2 3156070
3 3193075
4 3114614
但是,当我将相同的选择传递给 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)
解决方案
对于遇到此问题的任何其他人。read_sql_table 似乎不支持这个用例(此时)。如果您传入一个 SQLAlchemy Select 对象,它最终会被包装在另一个 SQLAlchemy Select 中并且没有别名,这是不好的 SQL(至少对于 PostgreSQL 而言)。
从 dask 源查看 read_sql_table , table 是传递给 read_sql_table 的 Select 对象,如所见,它被包装在另一个 select 中。
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)
推荐阅读
- javascript - 无法在 Contact Form 7 Plugin 上停止带有锚标记的表单提交
- c# - 无法得到我想要的字符串?
- hadoop - shell脚本中beeline命令的返回码
- vba - 强制性数据输入框
- java - 在单元测试期间找到资源文本文件,但在从 Netbeans 运行程序时未找到
- wordpress - 无法连接到本地 wordpress 服务器
- c# - NLog 未写入事件日志 .NET Core 2.1
- javascript - JavaScript 函数存在但未被调用
- sql - SQL 存储过程超时
- django - 将 React 流量重定向到开发中的 Django REST