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