python - 使用 pyODBC 的 fast_executemany 加速 pandas.DataFrame.to_sql

Question

我想向运行 MS SQL 的远程服务器发送一个大型 pandas.DataFrame。我现在这样做的方法是将 data_frame 对象转换为元组列表，然后使用 pyODBC 的 executemany() 函数将其发送出去。它是这样的:

 import pyodbc as pdb

 list_of_tuples = convert_df(data_frame)

 connection = pdb.connect(cnxn_str)

 cursor = connection.cursor()
 cursor.fast_executemany = True
 cursor.executemany(sql_statement, list_of_tuples)
 connection.commit()

 cursor.close()
 connection.close()

然后我开始怀疑是否可以通过使用 data_frame.to_sql() 方法来加快速度(或者至少更具可读性)。我想出了以下解决方案:

 import sqlalchemy as sa

 engine = sa.create_engine("mssql+pyodbc:///?odbc_connect=%s" % cnxn_str)
 data_frame.to_sql(table_name, engine, index=False)

现在代码更具可读性，但上传速度至少慢了 150 倍...

在使用 SQLAlchemy 时有没有办法翻转 fast_executemany？

我正在使用 pandas-0.20.3、pyODBC-4.0.21 和 sqlalchemy-1.1.13。

Answer 1

编辑(2019-03-08): Gord Thompson 在下面评论了来自 sqlalchemy 更新日志的好消息:自 2019-03-04 发布的 SQLAlchemy 1.3.0 以来，sqlalchemy现在支持 mssql+pyodbc 方言的 engine = create_engine(sqlalchemy_url, fast_executemany=True)。即，不再需要定义函数并使用 @event.listens_for(engine, 'before_cursor_execute') 这意味着可以删除以下函数，只需要设置标志在 create_engine 语句中 - 仍然保持加速。

原帖:

刚刚注册了一个帐户来发布这个。我想在上面的帖子下发表评论，因为它是对已经提供的答案的跟进。上面的解决方案适用于我在基于 Ubuntu 的安装中写入的 Microsft SQL 存储上的版本 17 SQL 驱动程序。

我用来显着加快速度的完整代码(加速 > 100 倍)如下。这是一个交 key 代码段，前提是您使用相关详细信息更改连接字符串。对于上面的海报，非常感谢您的解决方案，因为我已经为此寻找了相当长的时间。

import pandas as pd
import numpy as np
import time
from sqlalchemy import create_engine, event
from urllib.parse import quote_plus


conn =  "DRIVER={ODBC Driver 17 for SQL Server};SERVER=IP_ADDRESS;DATABASE=DataLake;UID=USER;PWD=PASS"
quoted = quote_plus(conn)
new_con = 'mssql+pyodbc:///?odbc_connect={}'.format(quoted)
engine = create_engine(new_con)


@event.listens_for(engine, 'before_cursor_execute')
def receive_before_cursor_execute(conn, cursor, statement, params, context, executemany):
    print("FUNC call")
    if executemany:
        cursor.fast_executemany = True


table_name = 'fast_executemany_test'
df = pd.DataFrame(np.random.random((10**4, 100)))


s = time.time()
df.to_sql(table_name, engine, if_exists = 'replace', chunksize = None)
print(time.time() - s)

根据下面的评论，我想花一些时间来解释一下 pandas to_sql 实现和查询处理方式的一些限制。有两件事可能会导致 MemoryError 被引发 afaik:

1) 假设您正在写入远程 SQL 存储。当您尝试使用 to_sql 方法编写 Pandas 数据帧时，它会将整个数据帧转换为值列表。这种转换比原始 DataFrame 占用更多的 RAM(最重要的是，旧的 DataFrame 仍然存在于 RAM 中)。此列表提供给您的 ODBC 连接器的最终 executemany 调用。我认为 ODBC 连接器在处理如此大的查询时遇到了一些麻烦。解决此问题的一种方法是为 to_sql 方法提供一个 chunksize 参数(10**5 似乎是最佳的，在 2 CPU 7GB ram MSSQL 存储上提供大约 600 mbit/s (!) 写入速度来自 Azure 的应用程序 - 不能推荐 Azure 顺便说一句)。因此，第一个限制，即查询大小，可以通过提供 chunksize 参数来规避。但是，这不会让您编写大小为 10**7 或更大的数据帧(至少在我正在使用的具有 ~55GB RAM 的 VM 上不是)，问题 nr 2。

这可以通过用 np.split (即 10**6 大小的 DataFrame block )分解 DataFrame 来规避。这些可以迭代地写掉。当我为 pandas 核心中的 to_sql 方法准备好解决方案时，我将尝试发出拉取请求，这样您就不必每次都进行预先分解。无论如何，我最终编写了一个与以下类似(不是交 key )的功能:

import pandas as pd
import numpy as np

def write_df_to_sql(df, **kwargs):
    chunks = np.split(df, df.shape()[0] / 10**6)
    for chunk in chunks:
        chunk.to_sql(**kwargs)
    return True

可以在此处查看上述代码段的更完整示例:https://gitlab.com/timelord/timelord/blob/master/timelord/utils/connector.py

这是我编写的一个类，它包含了补丁，并减轻了与 SQL 建立连接所带来的一些必要开销。还是要写一些文档。我还计划将补丁贡献给 pandas 本身，但还没有找到一个好的方法。

我希望这会有所帮助。