【问题标题】：Python/Psycopg2/PostgreSQL Copy_From loop gets slower as it progressesPython/Psycopg2/PostgreSQL Copy_From 循环随着它的进行而变慢
【发布时间】：2012-08-20 13:54:24
【问题描述】：

我编写了一个 Python 脚本，它采用 1.5 G XML 文件，解析数据并使用 copy_from 将其提供给数据库。它每 1000 个解析的节点调用以下函数。总共有大约 170k 个节点更新大约 300k 行或更多。它开始时非常快，然后随着时间的推移逐渐变慢。关于为什么会发生这种情况以及我能做些什么来解决它的任何想法？

这是我将数据提供给数据库的函数。

 def db_update(val_str, tbl, cols):

    conn = psycopg2.connect("dbname=<mydb> user=postgres password=<mypw>")
    cur = conn.cursor()
    output = cStringIO.StringIO()
    output.write(val_str)
    output.seek(0)
    cur.copy_from(output, tbl, sep='\t', columns=(cols))
    conn.commit()

我没有包含 xml 解析，因为我认为这不是问题。如果没有 db，解析器将在 2 分钟内执行。

【问题讨论】：

您是否尝试过缓存连接？您还可以通过将val_str 直接传递给StringIO() 来稍微简化函数，从而消除write 和随后的seek。
我是 StringIO() 新手。你能举例说明你的意思吗？可能很愚蠢，但如果我不问......
output = cStringIO.StringIO(val_str) 是获取包含val_str（适用于copy_from）的只读文件类对象所需的全部操作。基本上，如果你不给它任何参数，你会得到一个读/写文件，但是如果你给它一个字符串参数 cStringIO.StringIO 会给你一个带有指定内容的只读文件。见cStringIO.StringIO。

标签： python database postgresql psycopg2

【解决方案1】：

随着表的增长，有几件事会降低插入速度：

随着数据库的增长需要做更多工作的触发器
索引，随着它们的增长，更新成本会更高

禁用任何非关键触发器，或者如果不可能，则重新设计它们以在恒定时间内运行。

删除索引，然后在加载数据后创建它们。如果您需要实际 INSERTs 或 UPDATEs 的任何索引，则需要保留它们的成本。

如果您正在执行大量UPDATEs，请考虑定期VACUUMing 表，或将 autovacuum 设置为非常积极地运行。这将有助于 Pg 重用空间，而不是从文件系统中更昂贵地分配新空间，并且有助于避免表膨胀。

您还可以通过不为每个工作块重新连接来节省时间。保持连接。

【讨论】：

这一切都非常有帮助。很多这些技巧似乎都是针对一般优化的。我很好奇你认为最有可能导致性能急剧下降的罪魁祸首。
我猜你是在上面说的。我没有任何触发器。主键上的一个索引。
@MikeGirard 我没想到会有这样的下降，但是如果没有像 iostat 和 vmstat 测量、硬计时、Pg 日志等数据，很难说更多。事实上@ 987654327@ 没有提供任何有用的EXPLAIN ANALYZE 数据没有帮助。

【解决方案2】：

根据个人经验，copy_from 不会在你提交任何东西后更新任何索引，所以你必须稍后再做。我会将您的 conn = psycopg2.connect("dbname=<mydb> user=postgres password=<mypw>"); cur = conn.cursor() 移到函数之外，并在您完成插入所有内容后执行 commit()（我建议每 ~ 100k 行提交一次，否则它会开始变慢）。

此外，它可能看起来很愚蠢，但它经常发生在我身上：确保在调用 db_update 之后重置 val_str。对我来说，当 copy_from /inserts 开始变慢时，这是因为我插入了相同的行加上更多的行。

【讨论】：

【解决方案3】：

据我所见，我使用以下内容并没有对性能造成任何影响：

import psycopg2
import psycopg2.extras

local_conn_string = """
    host='localhost'
    port='5432'
    dbname='backupdata'
    user='postgres'
    password='123'"""
local_conn = psycopg2.connect(local_conn_string)
local_cursor = local_conn.cursor(
    'cursor_unique_name',
     cursor_factory=psycopg2.extras.DictCursor)

我在代码中生成了以下输出来测试运行时（我正在解析很多行。超过 30.000.000）。

Parsed 2600000 rows in 00:25:21
Parsed 2700000 rows in 00:26:19
Parsed 2800000 rows in 00:27:16
Parsed 2900000 rows in 00:28:15
Parsed 3000000 rows in 00:29:13
Parsed 3100000 rows in 00:30:11

我必须提到我不会“复制”任何东西。但是我正在将我的行从远程 PostGreSQL 移动到本地行，并且在此过程中创建了更多表来更好地索引我的数据，因为 30.000.000+ 在常规查询中处理起来有点太多了。

注意：时间是向上计数的，不是针对每个查询的。

我认为这与我的 cursor 的创建方式有关。

编辑1：

我正在使用以下内容来运行我的查询：

local_cursor.execute("""SELECT * FROM data;""")

row_count = 0
for row in local_cursor:
    if(row_count % 100000 == 0 and row_count != 0):
        print("Parsed %s rows in %s" % (row_count,
                                        my_timer.get_time_hhmmss()
                                        ))
    parse_row(row)
    row_count += 1

print("Finished running script!")
print("Parsed %s rows" % row_count)

my_timer 是我创建的计时器类，parse_row(row) 函数格式化我的数据，将其传输到我的本地数据库，并在验证数据已移动到我的数据后最终从远程数据库中删除本地数据库。

编辑2：

在我的数据库中每 100.000 行解析大约需要 1 分钟，即使在解析了大约 4.000.000 个查询之后：

Parsed 3800000 rows in 00:36:56
Parsed 3900000 rows in 00:37:54
Parsed 4000000 rows in 00:38:52
Parsed 4100000 rows in 00:39:50

【讨论】：