python - 如何优化我的代码以加快处理速度？

Question

我编写的代码存在一些性能问题。该代码的目的是比较 2 个 csv 文件(一个文件超过 900k 行，另一个文件有 50k ~ 80k 行)。

目标是比较 csv1 和 csv2，并将匹配数据写入第三个 csv。

我的数据是这样的:

CSV1:

address,name,order_no
add1,John,d009
add2,Smith,d019
add3,Mary,d890
.....(900k more rows)

CSV2:

address,hub_id
add3,x345
add4,x310
add1,a109
....(50k ~ 80k more rows)

预期输出:

CSV3:

order_no,hub_id
d890,x345
d009,a109
.....(etc)

我现在正在处理的代码(尽管很简单)确实有效。但是，整个比较和写入的过程需要很长时间才能完成。

任何指针将不胜感激。我可能忽略了一些python函数，可以在比较大数据的情况下使用，因为我刚刚开始学习。

import csv
import time
start_time = time.time()

with open('csv1.csv', newline='', encoding='Latin-1') as masterfile:
    reader = csv.DictReader(masterfile)
    for row in reader:
        with open('csv2.csv', newline='', encoding='Latin-1') as list1:
            reader2 = csv.DictReader(list1)
            for row2 in reader2:
                if row2['address'] == row['address']:
                     with open('csv3.csv', 'a') as corder:
                     print(row2['wip'] + ', ' + row['lat'] + ', ' + row['long'], file=corder)

print("--- %s seconds ---" % (time.time() - start_time))

Answer 1

你的算法目前正在做什么:

1. 加载一行大文件。
2. 打开较小的文件。
3. 在磁盘中的小文件中进行线性搜索
4. 打开输出文件并写入。
5. 冲洗并重复。

所有这些步骤都完成了 90 万次以上。

第 2 步，打开较小的文件，应该只执行一次。打开文件并从磁盘加载它是一项开销很大的操作。只需在开始时加载一次并在内存中进行线性搜索(步骤 #3)，您就会看到很大的改进。

步骤 #4 也是如此:打开输出文件应该只进行一次。每次关闭时，系统都会将文件刷新到磁盘。这是非常浪费的一步。如果您保持文件打开，输出缓冲数据，直到有足够的数据将一个完整的 block 写入磁盘，这是完成此操作的一种更快的方法。

步骤 #3 可以通过使用正确的数据结构进行大量优化。概率在日常生活中最常见的用途之一是哈希表。它们无处不在，因为它们使查找成为一个恒定时间的操作(与线性搜索不同，线性搜索随输入的大小线性缩放)。哈希表在 Python 的 dict 类中实现。通过创建以 address 为键的 dict，您可以将处理时间减少到 900k + 80k 的倍数而不是 900k * 80k。查找algorithmic complexity了解更多。我特别推荐 Steve Skiena 的《算法设计手册》。

最后一步是找到每个文件中地址的交集。有几个选项可用。您可以将两个文件都转换为 dict 并执行类似 set 的键交集，或者您可以将一个文件加载到 dict 中并逐行测试另一个。我强烈推荐后者，将较小的文件作为加载到 dict 中的文件。从算法的角度来看，元素减少 10 倍意味着您可以降低哈希冲突的可能性。这也是成本最低的方法，因为它在较大文件的不相关行上快速失败，而不记录它们。从实际的角度来看，您甚至可能无法直接将较大的文件转换为字典，如果我怀疑它有多个具有相同地址的行。

这是我一直在谈论的内容的实现:

with open('csv2.csv', newline='', encoding='Latin-1') as lookupfile:
    lookup = dict(csv.reader(lookupfile))

with open('csv1.csv', newline='', encoding='Latin-1') as masterfile, open('csv3.csv', 'w') as corder:
    reader = csv.reader(masterfile)
    corder.write('order_no,hub_id\n')
    for address, name, order_no in reader:
        hub_id = lookup.get(address)
        if hub_id is not None:
            corder.write(f'{order_no},{hub_id}\n')

如果任何行的长度不正好是两个元素，则表达式 dict(csv.reader(lookupfile)) 将失败。例如，空行会使它崩溃。这是因为 dict 的构造函数需要一个可迭代的双元素序列来初始化键值映射。

作为次要优化，我没有使用 csv.DictReader，因为这需要对每一行进行额外处理。此外，我已经从输出中完全删除了 csv 模块，因为您可以在不添加包装器层的情况下更快地完成工作。如果您的文件格式与您显示的一样整齐，您可能会通过将它们拆分为 , 自己而不是使用 csv 来获得微小的性能提升。