ASYNC - Pandas read_sql 和 asyncio？答案

【问题标题】：ASYNC - Pandas read_sql and asyncio?ASYNC - Pandas read_sql 和 asyncio？
【发布时间】：2018-07-19 15:27:12
【问题描述】：

有人可以为我指出如何解决以下问题的正确方向。我正在尝试使用 pandas.read_sql 和 asyncio 提出解决方案。我想将表记录从一个数据库迁移到另一个数据库。

我想做以下事情：

table 1
.
.
.
table n

我有这个功能：

def extract(table):
    try:
        df = pd.DataFrame()
        df = pd.concat(
              [chunk for chunk in
                  pd.read_sql(sql,
                              con=CONNECTION,
                              chunksize=10**5)]
                    )
    except Exception as e:
        raise e
    else:
        return df

我想并行运行这些，而不是一个一个地运行。

extract(table1)
extract(table2)
.
.
extract(tablen)

【问题讨论】：

asyncio 是硬性要求吗？您是否考虑过线程或多处理？
是的，但也许我可以使用线程或多处理来获得一些想法。但我听说使用这些方法会出现很多问题。
即使 asyncio 是一项硬性要求，基于 asyncio 的解决方案仍会在后台使用线程来并行运行 DataFrame.read_sql。考虑到这一点，最好使用concurrent.futures，它为并行化代码提供了出色的工具。

标签： python pandas python-asyncio python-3.7

【解决方案1】：

asyncio 是将非阻塞代码组织成回调和协程。并行运行 CPU 密集型代码是线程的一个用例：

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor() as executor:
    frames = list(executor.map(extract, all_tables))

这实际上是否会比顺序代码运行得更快取决于pd.read_sql 是否发布了GIL。

【讨论】：

有没有办法在python代码中检查或发布它？类似于：获取时：释放 GIL 以让其他函数并行运行？
另一个问题：您可以立即在 ThreadPoolExecutor 上管理结果还是需要等到所有提取（tab1...tabn）完成？
@Maki 您不能在 Python 中发布 GIL，但 C 扩展可以在安全的情况下这样做。（Panda 的作者是 aware 。）如果您需要在结果到达时对其进行管理，请查看 executor 的 submit 方法。它返回一个Future，您可以通过各种方式对其进行管理，包括注册一个回调以在结果准备好时执行。
@Maki 另见as_completed，一个接受一堆期货并在它们完成时生成它们的迭代器。
您可以使用 ProcessPoolExecutor 作为多处理的替代品。但是您可能会遇到将连接对象传递给子进程的问题，在这种情况下，您需要在那里打开一个新对象。