remove parallel blocking spike
@@ -113,21 +113,6 @@ class IntegralMatching(Matching):
return matches
-class BlockingProcess:
- def __init__(self, db_location, blocker, dtype, shape):
- self.db_location = db_location
- self.blocker = blocker
- self.data = numpy.memmap('foomap', dtype=dtype, shape=shape)
-
- def __call__(self, offset):
- data = [pickle.loads(row) for row in self.data[offset::4]]
- con = sqlite3.connect(self.db_location)
- con.executemany("INSERT INTO blocking_map values (?, ?)",
- self.blocker(data))
- con.commit()
-
-
-
class DedupeMatching(IntegralMatching):
"""
@@ -251,36 +236,8 @@ class DedupeMatching(IntegralMatching):
(block_key text, record_id {id_type})
'''.format(id_type=id_type))
- from multiprocessing import Pool
- import pickle
-
- pool = Pool(4)
- npdata = numpy.array([pickle.dumps((k, d)) for k, d in data.items()])
-
- fp = numpy.memmap('foomap', dtype=npdata.dtype, mode='w+', shape=npdata.shape)
- fp[:] = npdata[:]
- fp.flush()
-
- import time
- start = time.perf_counter()
- parallel = True
- if parallel:
-
-
- pool.map_async(BlockingProcess(temp_dir + '/blocks.db',
- self.fingerprinter.__call__,
- fp.dtype,
- fp.shape,
- ),
- (0, 1, 2, 3))
-
- pool.close()
- pool.join()
- else:
con.executemany("INSERT INTO blocking_map values (?, ?)",
self.fingerprinter(data.items()))
- end = time.perf_counter()
- print(end-start, 'time to block')
self.fingerprinter.reset_indices()
