我正在嘗試將 SQLAlchemy 表達式與 dask 的 read_sql_table 一起使用，以便關(guān)閉通過連接和過濾幾個不同表創(chuàng)建的數(shù)據(jù)集。該文件表明，這應(yīng)該是可能的。（下面的示例不包括任何連接，因為復(fù)制問題不需要它們。）我構(gòu)建了我的連接字符串，創(chuàng)建了一個 SQLAlchemy 引擎和與我的數(shù)據(jù)庫中的表相對應(yīng)的表。（我正在使用 PostgreSQL。）import dask.dataframe as ddimport pandas as pdfrom sqlalchemy import create_enginefrom sqlalchemy import Column, MetaData, Tablefrom sqlalchemy.sql import selectusername = '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')我能夠構(gòu)建一個選擇并將它與 SQLAlchemy 一起使用，沒有問題>>> s = select([t]).limit(5)>>> rp = engine.execute(s)>>> rp.fetchall()[(3140757,), (3118225,), (3156070,), (3193075,), (3114614,)]我還可以將 SQLAlchey 選擇提供給熊貓的 read_sql，它工作正常>>> pd.read_sql(s, connection_string)id0 31407571 31182252 31560703 31930754 3114614但是，當(dāng)我將相同的選擇傳遞給 dask 時，我收到了 ProgrammingError。它表明dask正在轉(zhuǎn)身并調(diào)用pandas.read_sql，所以你會認為它應(yīng)該工作，但顯然不是。
3 回答
白豬掌柜的
TA貢獻1893條經(jīng)驗 獲得超10個贊
正如 Chris 在不同的答案中所說，Dask 以某種形式包裝您的查詢SELECT columns FROM (yourquery)，這對于 PostgreSQL 來說是無效的語法，因為它需要括號表達式的別名。無需重新實現(xiàn)整個read_sql_table方法，表達式可以簡單地通過添加.alias('somename')到您的選擇中來別名，即
select([t]).limit(5).alias('foo')該表達式，當(dāng)被 Dask 包裝時，為 Postgres 生成正確的語法
SELECT columns FROM (yourquery) AS foo
慕婉清6462132
TA貢獻1804條經(jīng)驗 獲得超2個贊
對于遇到此問題的任何其他人。read_sql_table 似乎不支持這個用例（此時）。如果你傳入一個 SQLAlchemy Select 對象，它最終會被包裹在另一個 SQLAlchemy Select 中并且沒有別名，這是糟糕的 SQL（至少對于 PostgreSQL）。
從 dask 源查看 read_sql_table，table 是傳遞給 read_sql_table 的 Select 對象，正如所見，它被包裝在另一個選擇中。
q = sql.select(columns).where(sql.and_(index >= lower, cond)
).select_from(table)
好消息是 read_sql_table 函數(shù)相對簡單，而且魔術(shù)實際上只有幾行從延遲對象創(chuàng)建數(shù)據(jù)幀。您只需要編寫自己的邏輯即可將查詢分成塊
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)
躍然一笑
TA貢獻1826條經(jīng)驗 獲得超6個贊
該行發(fā)送的查詢是由 SQLAlchemy 自動生成的，因此語法應(yīng)該是正確的。但是，我注意到您的原始查詢包含一個.limit()修飾符。該行的目的head =是獲取前幾行，以推斷類型。如果原始查詢已經(jīng)有一個限制子句，我可以看到兩者可能會發(fā)生沖突。請嘗試使用不帶 的查詢.limit()。
添加回答
舉報
0/150
提交
取消