使用 psycopg2 将 pandas DataFrame 快速插入 Postgres DB

问题描述 投票:0回答:2

我正在尝试以最有效的方式(使用 Python 2.7)将 pandas DataFrame 插入 Postgresql DB (9.1)。
使用“cursor.execute_many”确实很慢,“DataFrame.to_csv(buffer,...)”和“copy_from”也很慢。
我发现已经很多了!网络上更快的解决方案(http://eatthedots.blogspot.de/2008/08/faking-read-support-for-psycopgs.html)我将其改编为与 pandas 一起使用。
我的代码可以在下面找到。
我的问题是这个相关问题的方法(使用“使用二进制从标准输入复制”)是否可以轻松地转移到使用 DataFrames 以及这是否会更快。
将二进制 COPY 表 FROM 与 psycopg2 一起使用
不幸的是,我的 Python 技能不足以理解这种方法的实现。
这是我的方法:


import psycopg2
import connectDB # this is simply a module that returns a connection to the db
from datetime import datetime

class ReadFaker:
    """
    This could be extended to include the index column optionally. Right now the index
    is not inserted
    """
    def __init__(self, data):
        self.iter = data.itertuples()

    def readline(self, size=None):
        try:
            line = self.iter.next()[1:]  # element 0 is the index
            row = '\t'.join(x.encode('utf8') if isinstance(x, unicode) else str(x) for x in line) + '\n'
        # in my case all strings in line are unicode objects.
        except StopIteration:
            return ''
        else:
            return row

    read = readline

def insert(df, table, con=None, columns = None):

    time1 = datetime.now()
    close_con = False
    if not con:
        try:
            con = connectDB.getCon()   ###dbLoader returns a connection with my settings
            close_con = True
        except psycopg2.Error, e:
            print e.pgerror
            print e.pgcode
            return "failed"
    inserted_rows = df.shape[0]
    data = ReadFaker(df)

    try:
        curs = con.cursor()
        print 'inserting %s entries into %s ...' % (inserted_rows, table)
        if columns is not None:
            curs.copy_from(data, table, null='nan', columns=[col for col in columns])
        else:
            curs.copy_from(data, table, null='nan')
        con.commit()
        curs.close()
        if close_con:
            con.close()
    except psycopg2.Error, e:
        print e.pgerror
        print e.pgcode
        con.rollback()
        if close_con:
            con.close()
        return "failed"

    time2 = datetime.now()
    print time2 - time1
    return inserted_rows
python postgresql psycopg2 pandas
2个回答
0
投票

Pandas 数据帧现在有一个 .to_sql 方法。目前尚不支持 Postgresql,但有一个补丁看起来可以使用。请参阅问题此处此处


-1
投票

我还没有测试性能,但也许你可以使用这样的东西:

  1. 遍历 DataFrame 的行,生成代表行的字符串(见下文)
  2. 在流中转换此可迭代对象,例如使用 Python:将可迭代对象转换为流?
  3. 最后在此流上使用 psycopg 的
    copy_from

要有效地生成 DataFrame 的行,请使用以下内容:

    def r(df):
            for idx, row in df.iterrows():
                    yield ','.join(map(str, row))
© www.soinside.com 2019 - 2024. All rights reserved.