如何在 python 中存储（和访问）一个非常大的列表

Question

我有一个包含超过 400 万行的文件，我想逐行读取，并对数据的第一个“列”执行操作。可能有重复的数据，我想确保我只执行一次这些操作，并且能够重新启动程序并从它停止的地方开始。

我的解决方案是解析该行，将值存储到一个列表中，然后“如果 x 不在列表中”执行操作，然后如果/当程序死亡/停止/被杀死时将该列表存储为泡菜。

但是脚本在运行几个小时后会出现段错误，我认为这是因为我在 checkFile 列表中存储了大量 MD5 和而烧毁了我的所有 ram。

因为文件太大（超过 400 万个条目），我需要一些快速查找的东西，希望有一个简单的解决方案。谢谢！

数据如下所示：6e5f6c90e6bf31c31283afb8dca93da8|mouse.gif|10102017

我当前的代码如下所示：

checkedFile = []

def readline(inFile):
print("Opening file: %s\n" % inFile)
try:
    with open(inFile, "r") as inputFile:
        for line in inputFile: # read in file line by line
            fileHash, garbage = line.split("|",1) 
            if fileHash not in checkedFile:
                checkedFile.append(fileHash)
                checkit(fileHash) # 

Answer 1

我想您遇到的问题之一是您选择的数据结构。由于checkedFile 是一个列表，因此检查一个哈希是否在列表中有O(N) time complexity。我的第一个建议是使用 python set。这使用了性能为O(1) 的哈希映射。

checkedFile = set()

def readline(inFile):
    print("Opening file: %s\n" % inFile)
    with open(inFile, "r") as inputFile:
        for line in inputFile: # read in file line by line
            fileHash, garbage = line.split("|",1) 
            if fileHash not in checkedFile:
                checkedFile.add(fileHash)
                checkit(fileHash)

Python 只为2 reasons 提供真正的段错误，因此您肯定会耗尽内存。我发现您使用列表存储进行分段错误很奇怪，因为每个哈希是 16 字节 * 400 万 = 64Mb，我认为这与您的典型计算机中的 ram 数量相差甚远。

我建议采用持久存储路线。这种方法不会耗尽内存，因为它会存储在您的硬盘上。

 import sqlite3
 connection = sqlite3.connect('cache.db')
 CREATE_CACHE_TABLE = '''CREATE TABLE IF NOT EXISTS cache ( 
                         hash TEXT PRIMARY KEY
               )'''
 SELECT_STATEMENT = 'SELECT hash FROM cache WHERE hash = ?'
 INSERT_STATEMENT = 'INSERT INTO cache VALUES ?'

 with connection:
     connection.execute(CREATE_CACHE_TABLE)
 with open(inFile, "r") as inputFile:
      for line in inputFile:
          fileHash, garbage = line.split("|", 1)
          with connection:
               if not connection.execute(SELECT_STATMENT, (fileHash,)).fetchone():
                    connection.execute(INSERT_STATEMENT, (fileHash,))
                    checkit(fileHash)

作为额外的奖励，您还可以缓存来自checkit 的结果，以便在由于某种原因停止计算时可以继续计算。这也将使您受益，因为 Web 请求将需要10ms 的最短响应时间。您提到 Web 请求返回一个 json 响应。我在sqlite with python 如何在sqlite 中存储json 之前写过这个。您需要修改函数checkit 以返回json 数据。

import sqlite3
import json

def adapt_json(data):
    return (json.dumps(data, sort_keys=True)).encode()

def convert_json(blob):
    return json.loads(blob.decode())

sqlite3.register_adapter(dict, adapt_json)
sqlite3.register_converter('JSON', convert_json)

connection = sqlite3.connect('cache.db', detect_types=sqlite3.PARSE_DECLTYPES)
CREATE_CACHE_TABLE = '''CREATE TABLE IF NOT EXISTS cache (
                         hash TEXT PRIMARY KEY,
                         check_results JSON
               )'''
SELECT_STATEMENT = 'SELECT hash FROM cache WHERE hash = ?'
INSERT_STATEMENT = 'INSERT INTO cache VALUES ?, ?'

with connection:
    connection.execute(CREATE_CACHE_TABLE)
with open(inFile, "r") as inputFile:
    for line in inputFile:
        fileHash, garbage = line.split("|", 1)
        with connection:
            if not connection.execute(SELECT_STATMENT, (fileHash,)).fetchone():
                json_data = checkit(fileHash)
                connection.execute(INSERT_STATEMENT, (fileHash, json_data))

如果您使用最后一种方法。您可以与文件的不同块并行运行此程序，它仍然可以工作。如果你最终得到一个更大的文件，这种方法应该可以很好地扩展。