比较两个大文件并合并匹配信息

Question

我有两个相当大的文件，JSON（185,000 行）和 CSV（650,000）。我需要遍历 JSON 文件中的每个 dict，然后在其中遍历 part_numbers 中的每个部分，并比较它以获取 CSV 中该部分所在位置的前三个字母。

由于某种原因，我很难正确地做到这一点。我的脚本的第一个版本太慢了，所以我正在尝试加快速度

JSON 示例：

[
    {"category": "Dryer Parts", "part_numbers": ["ABC", "DEF", "GHI", "JKL", "MNO", "PQR"], "parent_category": "Dryers"},
    {"category": "Washer Parts", "part_numbers": ["ABC", "DEF", "GHI", "JKL", "MNO", "PQR"], "parent_category": "Washers"},
    {"category": "Sink Parts", "part_numbers": ["ABC", "DEF", "GHI", "JKL", "MNO", "PQR"], "parent_category": "Sinks"},
    {"category": "Other Parts", "part_numbers": ["ABC", "DEF", "GHI", "JKL", "MNO", "PQR"], "parent_category": "Others"}
]

CSV：

WCI|ABC
WPL|DEF
BSH|GHI
WCI|JKL

结束的字典如下所示：

{"category": "Other Parts",
 "part_numbers": ["WCIABC","WPLDEF","BSHGHI","JKLWCI"...]}

这是我到目前为止所做的一个示例，它在if (part.rstrip() == row[1]): 返回IndexError: list index out of range：

import csv
import json
from multiprocessing import Pool

def find_part(item):
    data = {
        'parent_category': item['parent_category'],
        'category': item['category'],
        'part_numbers': []
    }

    for part in item['part_numbers']:
        for row in reader:
            if (part.rstrip() == row[1]):
                data['part_numbers'].append(row[0] + row[1])

    with open('output.json', 'a') as outfile:
        outfile.write('    ')
        json.dump(data, outfile)
        outfile.write(',\n')


if __name__ == '__main__':
    catparts = json.load(open('catparts.json', 'r'))
    partfile = open('partfile.csv', 'r')
    reader = csv.reader(partfile, delimiter='|')


    with open('output.json', 'w+') as outfile:
        outfile.write('[\n')

    p = Pool(50)
    p.map(find_part, catparts)

    with open('output.json', 'a') as outfile:
        outfile.write('\n]')

Answer 1

正如我在评论中所说，您的代码（现在）给了我一个 NameError: name 'reader' 未在 find_part() 函数中定义。解决方法是将csv.reader 的创建移到函数中。我还更改了文件的打开方式，以使用with 上下文管理器和newline 参数。这也解决了一堆单独的任务都试图同时读取同一个 csv 文件的问题。

您的方法效率非常低，因为它为item['part_numbers'] 中的每个部分读取整个'partfile.csv' 文件。不过，以下似乎可行：

import csv
import json
from multiprocessing import Pool

def find_part(item):
    data = {
        'parent_category': item['parent_category'],
        'category': item['category'],
        'part_numbers': []
    }

    for part in item['part_numbers']:
        with open('partfile.csv', newline='') as partfile:  # open csv in Py 3.x
            for row in csv.reader(partfile, delimiter='|'):
                if part.rstrip() == row[1]:
                    data['part_numbers'].append(row[0] + row[1])

    with open('output.json', 'a') as outfile:
        outfile.write('    ')
        json.dump(data, outfile)
        outfile.write(',\n')

if __name__ == '__main__':
    catparts = json.load(open('carparts.json', 'r'))

    with open('output.json', 'w+') as outfile:
        outfile.write('[\n')

    p = Pool(50)
    p.map(find_part, catparts)

    with open('output.json', 'a') as outfile:
        outfile.write(']')

这是一个更高效的版本，每个子进程仅读取整个 'partfile.csv' 文件一次：

import csv
import json
from multiprocessing import Pool

def find_part(item):
    data = {
        'parent_category': item['parent_category'],
        'category': item['category'],
        'part_numbers': []
    }

    with open('partfile.csv', newline='') as partfile:  # open csv for reading in Py 3.x
        partlist = [row for row in csv.reader(partfile, delimiter='|')]

    for part in item['part_numbers']:
        part = part.rstrip()
        for row in partlist:
            if row[1] == part:
                data['part_numbers'].append(row[0] + row[1])

    with open('output.json', 'a') as outfile:
        outfile.write('    ')
        json.dump(data, outfile)
        outfile.write(',\n')

if __name__ == '__main__':
    catparts = json.load(open('carparts.json', 'r'))

    with open('output.json', 'w+') as outfile:
        outfile.write('[\n')

    p = Pool(50)
    p.map(find_part, catparts)

    with open('output.json', 'a') as outfile:
        outfile.write(']')

虽然您可以在主任务中将'partfile.csv' 数据读入内存并将其作为参数传递给find_part() 子任务，但这样做只是意味着必须对每个进程的数据进行腌制和解封。您需要运行一些计时测试来确定这是否比使用csv 模块显式读取它更快，如上所示。

还要注意，在将任务提交到Pool 之前，预处理来自'carparts.json' 文件的数据加载并从每行的第一个元素中去除尾随空格也会更有效，因为这样您就不需要一遍又一遍地在find_part() 中执行part = part.rstrip()。同样，我不知道这样做是否值得努力——只有时间测试才能确定答案。

Answer 2

我想我找到了。您的 CSV 阅读器与许多其他文件访问方法一样：您按顺序读取文件，然后点击 EOF。当您尝试对第二部分执行相同操作时，该文件已经在 EOF，并且第一次 read 尝试返回 null 结果；这没有第二个元素。

如果您想再次访问所有记录，您需要重置文件书签。最简单的方法是使用

回溯到字节 0

partfile.seek(0)

另一种方法是关闭并重新打开文件。

这会让你感动吗？

Answer 3

只要所有部件号都存在于 csv 中，这应该可以工作。

import json

# read part codes into a dictionary
with open('partfile.csv') as fp:
    partcodes = {}
    for line in fp:
        code, number = line.strip().split('|')
        partcodes[number] = code

with open('catparts.json') as fp:
    catparts = json.load(fp)

# modify the part numbers/codes 
for cat in catparts:
    cat['part_numbers'] = [partcodes[n] + n for n in cat['part_numbers']]

# output
with open('output.json', 'w') as fp:
    json.dump(catparts, fp)