如何将大型 Oracle 表的 SUBSET 加载到 Dask 数据帧中？

Question

这是我尝试过的：

dask_rf = dd.from_pandas(pd.read_sql('select ...)', conn_cx_Oracle), npartitions = 10)

这给了我一个“大对象”警告并建议使用 client.scatter。问题是 client.scatter 似乎需要首先将数据加载到 Pandas 数据帧中，这就是为什么我首先使用 Dask 的原因，因为 RAM 限制。

Oracle 表太大，无法使用 Dask 的 read_sql_table 读取，因为 read_sql_table 不会以任何方式过滤该表。

想法？ Dask 不适用于我的用例？

编辑 - 根据下面的答案和研究如何这样做之后，这是我尝试转换为使用 sqlalchemy 表达式：

from sqlalchemy import create_engine, Table, Column, String, MetaData, select

sql_engine = create_engine(f'oracle+cx_oracle://username:password@environment')

metadata = MetaData(bind=sql_engine)

table_reference = Table('table', metadata, autoload=True, schema='schema')

s = select([table_reference ]).where(table_reference .c.field_to_filter == filtered_value)

import dask.dataframe as dd

dask_df = dd.read_sql_table(s, 'sqlalchemy_connection_string', 'index_col', schema = 'schema')

dask_df.count()

Dask 系列结构：npartitions=1 action_timestamp int64 vendor_name ... dtype：int64 Dask 名称：dataframe-count-agg， 1996年任务

dask_df.count().compute()

DatabaseError: (cx_Oracle.DatabaseError) ORA-02391: 超出 同时 SESSIONS_PER_USER 限制（此错误的背景： http://sqlalche.me/e/4xp6)

为什么要尝试连接到 Oracle？

编辑 #2 - 以防万一，我进行了额外的测试。我想证明 sqlalchemy 可以独立工作，所以我通过以下方式证明了这一点：

result = sql_engine.execute(s)

type(result)

sqlalchemy.engine.result.ResultProxy

result.fetchone()

结果已显示

这似乎排除了 SQLAlchemy/Oracle 问题，那么有什么想法可以尝试下一步吗？

Answer 1

我现在正在寻找同样的东西。

为了不被卡住...您可能没有足够的 RAM，但您可能有很多可用存储空间。所以...现在的建议

# imports
import pandas as pd 
import cx_Oracle as cx
import dask.dataframe as dd

# Connection stuff
...
conn = ...

# Query
qry = "SELECT * FROM HUGE_TABLE"

# Pandas Chunks
for ix , chunk in enumerate(pd.io.sql.read_sql(qry , conn , ... , chunksize=1000000)):
    pd.DataFrame(chunk).to_csv(f"chunk_{ix}.csv" , sep=";") # or to_parquet 

# Dask dataframe reading from files (chunks)
dataset = dd.read_csv("chunk_*.csv" , sep=";" , blocksie=32e6) # or read_parquet

由于这是 IO 密集型操作，并且您正在执行顺序操作，因此可能需要一段时间。

我对“导出”更快的建议是对表进行分区并按每个分区并行执行块导出。

Answer 2

问题在于 client.scatter 似乎需要先将数据加载到 Pandas 数据帧中

那是因为你在这里调用 Pandas 代码

dd.from_pandas(pd.read_sql('select ...)', conn_cx_Oracle), npartitions = 10)
#              pd.read_sql('select ...)', conn_cx_Oracle) # <<-----

根据read_sql_table docstring 你应该能够传入一个 SQLAlchemy 表达式对象。

read_sql_table 不接受任意 SQL 查询字符串的原因是因为它需要能够对您的查询进行分区，因此每个任务只加载整体的一部分。对于那里的许多方言来说，这是一件棘手的事情，所以我们依靠 sqlalchemy 来进行格式化。