2つの大きなファイルを比較し、一致する情報を組み合わせる

Question

JSON（185,000行）とCSV（650,000）の2つのファイルがあります。私はpart_numbersの各部分を反復処理し、それを比較して、その部分がCSVで見つかった場所から最初の3文字を取得することがJSONファイル内の各辞書を反復する必要があります。

何らかの理由で、私はこれを正しく行うのに苦労しています。私のスクリプトの最初のバージョンはあまりにも遅かったので、私は

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 

エンドdictのは次のようになりますが以下の下：

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

ここで私はこれまでのところ、それはで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ファイルを同時に読み込もうとする別々のタスクの問題を解決します。

のすべての部分について'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'ファイル読み込みはるかに効率的なバージョンです：あなたはメモリに'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(']') 

をメインタスクではサブタスクfind_part()に引数として渡します。これを行うと、すべてのプロセスでデータをpickle化してunpickleする必要があります。上記のように、csvモジュールを明示的に読み取るよりも高速であるかどうかを判断するには、いくつかのタイミングテストを実行する必要があります。

はまた、あなたがpart = part.rstrip()を行う必要がありませんので、それはまた'carparts.json'ファイルからのデータのロード前処理に、より効率的かつPoolにタスクを提出する前に、すべての行の最初のelemから末尾の空白を取り除くことに注意してくださいfind_part()に何度も繰り返しています。繰り返しますが、そうすることが努力の価値があるかどうかはわかりません。タイミングテストだけで答えを判断することができます。

Answer 2

私はそれを見つけたと思う。あなたのCSVリーダーは他の多くのファイルアクセス方法と似ています。ファイルを順番に読み取った後、EOFを押します。 2番目の部分で同じ処理を実行しようとすると、ファイルはすでにEOFにあり、最初のreadの試行ではnullの結果が返されます。これには2番目の要素はありません。

すべてのレコードに再度アクセスする場合は、ファイルのブックマークをリセットする必要があります。最も簡単な方法は、バイト0に戻すことです。

partfile.seek(0) 

もう1つの方法は、ファイルを閉じて再度開くことです。

あなたは動いていますか？

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)