【问题标题】:python pandas to_sql with sqlalchemy : how to speed up exporting to MS SQL?python pandas to_sql with sqlalchemy:如何加快导出到 MS SQL 的速度?
【发布时间】:2015-06-24 17:20:26
【问题描述】:

我有一个包含 155,000 行和 12 列的数据框。 如果我使用 dataframe.to_csv 将其导出到 csv,则输出为 11MB 文件(即刻生成)。

但是,如果我使用 to_sql 方法导出到 Microsoft SQL Server,则需要 5 到 6 分钟! 没有列是文本:只有 int、float、bool 和日期。我见过 ODBC 驱动程序设置 nvarchar(max) 的情况,这会减慢数据传输速度,但这里不是这样。

关于如何加快导出过程有什么建议吗?导出 11 MB 数据需要 6 分钟,这使得 ODBC 连接实际上无法使用。

谢谢!

我的代码是:

import pandas as pd
from sqlalchemy import create_engine, MetaData, Table, select
ServerName = "myserver"
Database = "mydatabase"
TableName = "mytable"

engine = create_engine('mssql+pyodbc://' + ServerName + '/' + Database)
conn = engine.connect()

metadata = MetaData(conn)

my_data_frame.to_sql(TableName,engine)

【问题讨论】:

  • 我唯一能想到的就是将结构(即列名和数据类型但没有行)导出到 SQL,然后将文件导出到 CSV 并使用类似导入/导出向导的东西将 CSV 文件附加到 SQL 表。这样我就不必再次定义所有的列类型;这很重要,因为导入工具倾向于读取前 x 行来猜测数据类型,如果第一行都是 NULL,那么猜测将是错误的。然而,事实仍然是 to_sql 方法除了用于小型表之外实际上无法使用。您在其他数据库中也遇到过这种情况吗?
  • 我在家里尝试了同样的方法,在我的同一台 PC 上运行 SQL Server Express,python 花了 2 分钟将 100 万行 x 12 列随机数的数据帧传输到 SQL(大小为CSV = 228MB)。不是超级快但可以接受。在连接到几英里外的 SQL 服务器的工作 PC 上花费了 6 分钟(对于一个小得多的文件)。不知道pandas、sqlalchemy或者pyodbc里面有没有参数可以加快传输速度?我经常使用许多其他工具连接到同一个 SQL 服务器,而且速度从来没有那么慢。谢谢!
  • 有人吗?我还验证了 pandas.read_sql_table 方法相当快。即使在编写没有约束的表时,也只是编写速度很慢。有任何想法吗?我不可能是唯一一个经历过这种情况的人,但我似乎无法在网上找到任何关于此的文档...... :(
  • 也许尝试按块大小分解它?因此,例如,让 for 循环遍历 10,000 行的块 (my_data_frame.to_sql(TableName,engine,chunksize=10000))。
  • 或者只是将数据导出到 csv,然后使用批量插入(非常非常快)。您将不得不构建一个格式文件,但这可能是值得的。 link

标签: python sql pandas sqlalchemy pyodbc


【解决方案1】:

DataFrame.to_sql 方法为您的 ODBC 连接器生成插入语句,然后 ODBC 连接器将其视为常规插入。

当这很慢时,这不是 pandas 的错。

DataFrame.to_sql 方法的输出保存到文件中,然后通过 ODBC 连接器重放该文件将花费相同的时间。

将数据批量导入数据库的正确方法是生成一个 csv 文件,然后使用加载命令,在 SQL 数据库的 MS 风格中称为 BULK INSERT

例如:

BULK INSERT mydatabase.myschema.mytable
FROM 'mydatadump.csv';

语法参考如下:

BULK INSERT 
   [ database_name . [ schema_name ] . | schema_name . ] [ table_name | view_name ] 
      FROM 'data_file' 
     [ WITH 
    ( 
   [ [ , ] BATCHSIZE = batch_size ] 
   [ [ , ] CHECK_CONSTRAINTS ] 
   [ [ , ] CODEPAGE = { 'ACP' | 'OEM' | 'RAW' | 'code_page' } ] 
   [ [ , ] DATAFILETYPE = 
      { 'char' | 'native'| 'widechar' | 'widenative' } ] 
   [ [ , ] FIELDTERMINATOR = 'field_terminator' ] 
   [ [ , ] FIRSTROW = first_row ] 
   [ [ , ] FIRE_TRIGGERS ] 
   [ [ , ] FORMATFILE = 'format_file_path' ] 
   [ [ , ] KEEPIDENTITY ] 
   [ [ , ] KEEPNULLS ] 
   [ [ , ] KILOBYTES_PER_BATCH = kilobytes_per_batch ] 
   [ [ , ] LASTROW = last_row ] 
   [ [ , ] MAXERRORS = max_errors ] 
   [ [ , ] ORDER ( { column [ ASC | DESC ] } [ ,...n ] ) ] 
   [ [ , ] ROWS_PER_BATCH = rows_per_batch ] 
   [ [ , ] ROWTERMINATOR = 'row_terminator' ] 
   [ [ , ] TABLOCK ] 
   [ [ , ] ERRORFILE = 'file_name' ] 
    )] 

【讨论】:

【解决方案2】:

您可以使用具有fast pandas to SQL functionalityd6tstack,因为它使用本机数据库导入命令。它支持 MS SQL、Postgres 和 MYSQL

uri_psql = 'postgresql+psycopg2://usr:pwd@localhost/db'
d6tstack.utils.pd_to_psql(df, uri_psql, 'table')
uri_mssql = 'mssql+pymssql://usr:pwd@localhost/db'
d6tstack.utils.pd_to_mssql(df, uri_mssql, 'table', 'schema') # experimental

对于在写入数据库之前导入多个带有数据架构更改的 CSV 和/或使用 pandas 进行预处理也很有用,请参阅examples notebook 中的进一步内容

d6tstack.combine_csv.CombinerCSV(glob.glob('*.csv'), 
    apply_after_read=apply_fun).to_psql_combine(uri_psql, 'table')

【讨论】:

  • 连接字符串中的用户名和密码应该放在哪里?可以举个例子吗?
  • 谢谢!在一个 3000 万行的 pandas 数据帧上运行 pd_to_psql,大约需要 4 分钟。
【解决方案3】:

基于这个答案 - Aseem

您可以使用 copy_from 方法来模拟带有游标对象的批量加载。 这是在 Postgres 上测试的,用你的数据库试试吧:

import pandas as pd
from sqlalchemy import create_engine, MetaData, Table, select
from StringIO import StringIO

ServerName = "myserver"
Database = "mydatabase"
TableName = "mytable"

engine = create_engine('mssql+pyodbc://' + ServerName + '/' + Database) #don't forget to add a password if needed

my_data_frame.head(0).to_sql(TableName, engine, if_exists='replace', index=False)  # create an empty table - just for structure
conn = engine.raw_connection()
cur = conn.cursor()
output = StringIO()
my_data_frame.to_csv(output, sep='\t', header=False, index=False) # a CSV that will be used for the bulk load
output.seek(0)
cur.copy_from(output, TableName, null="")  # null values become ''
conn.commit()
conn.close()
cur.close()

【讨论】:

    【解决方案4】:

    如果这对任何人都有帮助,我对这个问题的解决方案如下。根据我的阅读,pandas tosql 方法一次加载一条记录。

    您可以创建一个加载 1000 行并提交该事务的批量插入语句,而不是每次都提交一行。这大大提高了速度。

    import pandas as pd
    from sqlalchemy import create_engine
    import pymssql
    import os
    
    connect_string  = [your connection string]
    engine = create_engine(connect_string,echo=False)
    connection = engine.raw_connection()
    cursor = connection.cursor()
    
    def load_data(report_name):
        # my report_name variable is also my sql server table name so I use that variable to create table name string
        sql_table_name = 'AR_'+str(report_name)
        global chunk # to QC chunks that fail for some reason
        for chunk in pd.read_csv(report_full_path_new,chunksize=1000):
            chunk.replace('\'','\'\'',inplace=True,regex=True) #replace single quotes in data with double single quotes to escape it in mysql
            chunk.fillna('NULL',inplace=True)
    
            my_data = str(chunk.to_records(index=False).tolist()) # convert data to string 
            my_data = my_data[1:-1] # clean up the ends
            my_data = my_data.replace('\"','\'').replace('\'NULL\'','NULL') #convert blanks to NULLS for mysql
            sql_table_name = [your sql server table name]
    
            sql = """
            INSERT INTO {0} 
            VALUES {1}
    
             """.format(sql_table_name,my_data)
    
            cursor.execute(sql)
            # you must call commit() to persist your data if you don't set autocommit to True
            connection.commit()
    

    【讨论】:

      【解决方案5】:

      我最近遇到了同样的问题,想为其他人添加一个答案。 to_sql 似乎为每一行发送一个 INSERT 查询,这使得它真的很慢。但是由于0.24.0pandas.to_sql() 中有一个method 参数,您可以在其中定义自己的插入函数,或者只使用method='multi' 告诉pandas 在单个INSERT 查询中传递多行,这使得它更快。

      请注意,您的数据库可能有参数限制。在这种情况下,您还必须定义一个块大小。

      所以解决方案应该如下所示:

      my_data_frame.to_sql(TableName, engine, chunksize=<yourParameterLimit>, method='multi')
      

      如果你不知道你的数据库参数限制,那就试试不带chunksize参数吧。它会运行或给你一个错误,告诉你你的限制。

      【讨论】:

      • 我在将数据加载到 Postgres 时包含了 method='multi' ,它的加载速度提高了 1000 倍 :) 900k 行的数据无法在 6 小时内完成。但是当我使用“multi”时,需要 5 分钟。感谢您的提示。
      • 相同。这就像一个魅力。有人可以在答案中解释块大小吗?
      • @technazi 正如我在回答中所说,一次发送所有行可能会超出您的数据库参数限制并导致错误。为避免这种情况,您可以指定 chunksize。这会将插入划分为您在 chunksize 中指定的行数的块。如果您的数据库有一个参数限制,例如100 000 并且您的 DataFrame 有 100 万行,除非您添加 chunksize=100000,否则它将失败。
      • 我正在执行的许多上传作业将完全超时。现在我添加了 method='multi' 并且脚本可以完美运行。非常感谢您的帖子!
      • 为什么这不是默认参数 ;-(
      【解决方案6】:

      你可以使用这个:让它更快的是 pandas to_sqlmethod 参数。我希望这个帮助会有所帮助。

      根据我的经验,这个结果是从无限时间到 8 秒。

      
      df = pd.read_csv('test.csv')
      
      conn = create_engine(<connection_string>)
      
      start_time = time.time()
      df.to_sql('table_name', conn, method='multi',index=False, if_exists='replace')
      print("--- %s seconds ---" % (time.time() - start_time))
      

      【讨论】:

        【解决方案7】:

        我的时间和内存不足(为从 120MB CSV 加载的 DataFrame 分配了超过 18GB):

        df.to_sql('my_table', engine, if_exists='replace', method='multi', dtype={"text_field": db.String(64), "text_field2": db.String(128), "intfield1": db.Integer(), "intfield2": db.Integer(), "floatfield": db.Float()})
        

        这是帮助我同时导入和跟踪插入进度的代码:

        import sqlalchemy as db
        engine = db.create_engine('mysql://user:password@localhost:3306/database_name', echo=False)
        connection = engine.connect()
        metadata = db.MetaData()
        
        my_table = db.Table('my_table', metadata,
                      db.Column('text_field', db.String(64), index=True),
                      db.Column('text_field2', db.String(128), index=True),
                      db.Column('intfield1', db.Integer()),
                      db.Column('intfield2', db.Integer()),
                      db.Column('floatfield', db.Float())
                     )
        metadata.create_all(engine)
        kw_dict = df.reset_index().sort_values(by="intfield2", ascending=False).to_dict(orient="records")
        
        batch_size=10000
        for batch_start in range(0, len(kw_dict), batch_size):
            print("Inserting {}-{}".format(batch_start, batch_start + batch_size))
            connection.execute(my_table.insert(), kw_dict[batch_start:batch_start + batch_size])
        

        【讨论】:

        • 您应该将chunksize 参数添加到pd.to_sql,例如5000。正如文档中所说的chunksize int, Specify the number of rows in each batch to be written at a time. By default, all rows will be written at once.
        【解决方案8】:

        为什么pandas.DataFrame.to_sql 很慢?

        将数据从pandas 上传到Microsoft SQL Server 时,实际上大部分时间都花在将pandas 转换为Python 对象到MS SQL ODBC 驱动程序所需的表示形式上。 pandas 的分析速度比基本 Python 代码快得多的原因之一是,它适用于整数/浮点数的精益原生数组……它们的开销与它们各自的 Python 对应物不同。 to_sql 方法实际上将所有这些精益列转换为许多单独的 Python 对象,因此没有像其他 pandas 操作那样得到通常的性能处理。

        使用turbodbc.Cursor.insertmanycolumns 加快速度

        给定pandas.DataFrame,您可以使用turbodbcpyarrow 插入数据,其转换开销比转换为Python 对象时更少。

        import pyarrow as pa
        import turbodbc
        
        cursor = …  # cursor to a MS SQL connection initiated with turbodbc
        df = …  # the pd.DataFrame to be inserted
        
        # Convert the pandas.DataFrame to a pyarrow.Table, most of the columns
        # will be zero-copy and thus this is quite fast.
        table = pa.Table.from_pandas(table)
        
        # Insert into the database
        cursor.executemanycolumns("INSERT INTO my_table VALUES (?, ?, ?)",
                                   table)
        

        为什么会更快?

        我们正在做一个转换路径pd.DataFrame -> pyarrow.Table -> ODBC 结构,而不是pd.DataFrame -> Python 对象集合-> ODBC 数据结构的转换。由于以下原因,这更高效:

        • pandas.DataFrame 的大部分列可以转换为pyarrow.Table 的列而无需复制。表的列将引用相同的内存。所以没有进行实际的转换。
        • 转换完全在具有本机类型的本机代码中完成。这意味着,只要我们没有 object 类型的列,我们就不会在任何阶段发生 Python 对象的开销。

        【讨论】:

          【解决方案9】:

          正如在其他答案中所说,减速和/或超时的原因是熊猫一遍又一遍地插入许多单行。大量插入命令速度很慢和/或可能使目标数据库过载。

          使用 method='multi' 告诉 pandas 分块上传。这要快得多,而且不会那么容易超时。

          sqlEngine=create_engine('mysql+mysqlconnector://'+config['user']+':'+config['pass']+'@'+config['host']+'/'+config['dbname'])
          dbConnection=sqlEngine.connect()
          df.to_sql('table_name',con=dbConnection,method='multi',if_exists='append',index=False)
          dbConnection.close()
          

          【讨论】:

            【解决方案10】:

            使用SQLAlchemy&gt;=1.3,在创建engine 对象时,设置fast_executemany=TrueReference

            【讨论】:

              猜你喜欢
              • 2014-09-26
              • 2017-05-24
              • 2015-09-27
              • 2019-07-05
              • 1970-01-01
              • 1970-01-01
              • 2019-05-22
              • 1970-01-01
              • 1970-01-01
              相关资源
              最近更新 更多