Dask + PostgreSQL read_sql_table：不正确的数据类型 index_col答案

【问题标题】：Dask + PostgreSQL read_sql_table: incorrect data type index_colDask + PostgreSQL read_sql_table：不正确的数据类型 index_col
【发布时间】：2020-01-01 13:15:35
【问题描述】：

我正在尝试在一个非常大的 PostgreSQL 表中的列上运行 mean() 函数。由于我无法将列加载到内存中，因此我选择了使用 Dask 分布式进行并行和分区。

系统配置：

12 cores / 24 threads
64 GB RAM
SSD

我已尝试 (1) 使用 1 个进程最大化线程数 (24)，并且 (2) 我尝试使用每个线程最大化进程数 (24)。

问题在于，无论哪种情况，from-delayed 任务都需要 >= 24 秒。此任务主要由用于设置 SQL 连接的 sqlalchemy 组件组成。

from-delayed任务的执行时间长似乎与sql查询有关：我的索引列是数据类型bigint，但Dask根据float（1000.01）产生where条件。这在整个操作中是一笔巨大的成本。尽管 Dask 数据帧指示 dtype int64，但数据本身实际上是浮点数。

read_sql_table：

data = dd.read_sql_table("<table>",'postgresql+psycopg2://<user>:<pw>@<ip>:<port>/<db>',index_col='<int_col>',bytes_per_chunk=1e6 - 1e9)

我希望：

select * from <table> where <int_col> >= int and <int_col> < int

但 Dask 会这样做：

select * from <table> where <int_col> >= float and <int_col> < float

为什么 Dask 不在 SQL 查询中应用正确的索引数据类型？

【问题讨论】：

标签： python postgresql dask

【解决方案1】：

https://github.com/dask/dask/blob/master/dask/dataframe/io/sql.py

解决办法，从sql.py第186行开始：

parts = []
lowers, uppers = divisions[:-1], divisions[1:]
for i, (lower, upper) in enumerate(zip(lowers, uppers)):
    if dtype.kind == "i":
        upper = int(np.ceil(upper))
        lower = int(np.ceil(lower))
    cond = index <= upper if i == len(lowers) - 1 else index < upper
    q = sql.select(columns).where(sql.and_(index >= lower, cond)).select_from(table)
    parts.append(delayed(_read_sql_chunk)(q, uri, meta, **kwargs))

【讨论】：