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

最后一步是查找每个文件中地址的交集。有几个选项可供选择。您可以将这两个文件转换为 dict s 并执行 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 .