(sets_files, num_perm, threshold, num_part, m, storage_config)
| 237 | # 将csv文件列表中每个表格中的column转换为三元组(minhashes, sets, keys)形式 |
| 238 | # @profile |
| 239 | def bootstrap_index_sets(sets_files, num_perm, threshold, num_part, m, storage_config): |
| 240 | # 从csv中读取加载 |
| 241 | print("Creating sets...") |
| 242 | print("Using num_perm = {}".format(num_perm)) |
| 243 | |
| 244 | # 处理完整路径 |
| 245 | # sets_files = [os.path.join(set_path, sets_file) for sets_file in sets_files] |
| 246 | start = time.perf_counter() |
| 247 | lsh = MinHashLSHEnsemble(threshold=threshold, num_perm=num_perm, |
| 248 | num_part=num_part, m=m, storage_config=storage_config) |
| 249 | |
| 250 | startr = time.perf_counter() |
| 251 | |
| 252 | # 读入候选集目录 |
| 253 | sub_sets_files = split_list(sets_files, split_num) |
| 254 | process_list = [] |
| 255 | # pocessing |
| 256 | pre_q = multiprocessing.Queue() |
| 257 | sizes = [] |
| 258 | for i in range(split_num): |
| 259 | process = multiprocessing.Process(target=pre_multi_process, args=(sub_sets_files[i], pre_q, i)) |
| 260 | process.daemon = True |
| 261 | process_list.append(process) |
| 262 | process.start() |
| 263 | for i in range(split_num): |
| 264 | res = pre_q.get() |
| 265 | sizes.extend(res[0]) |
| 266 | sys.stdout.write("\rRead and size {}/{} sets".format(i+1,split_num)) |
| 267 | sys.stdout.write("\n") |
| 268 | for i, process in enumerate(process_list): |
| 269 | process.join() |
| 270 | |
| 271 | # threading |
| 272 | # pre_q = queue.Queue() |
| 273 | # sizes = [] |
| 274 | # for i in range(split_num): |
| 275 | # process = threading.Thread(target=pre_multi_process, args=(set_path, sub_sets_files[i], pre_q)) |
| 276 | # process_list.append(process) |
| 277 | # process.start() |
| 278 | # for i in range(split_num): |
| 279 | # res = pre_q.get() |
| 280 | # sizes.extend(res[0]) |
| 281 | # for key in res[1]: |
| 282 | # name_to_id[key] = len(id_to_name) |
| 283 | # id_to_name.append(key) |
| 284 | # sys.stdout.write("\rRead and size {}/{} sets".format(i+1,split_num)) |
| 285 | # sys.stdout.write("\n") |
| 286 | # for i, process in enumerate(process_list): |
| 287 | # process.join() |
| 288 | |
| 289 | print("size time = {}".format(time.perf_counter() - startr)) |
| 290 | print("size num = {}".format(len(sizes))) |
| 291 | partitions = lsh.count_partition(sizes) |
| 292 | |
| 293 | entries_q = multiprocessing.Queue() |
| 294 | param_dic = {'threshold':threshold, 'num_perm':num_perm, 'num_part':num_part, 'm':m, 'storage_config':storage_config, "partitions":partitions} |
| 295 | for i in range(split_num): |
| 296 | process = multiprocessing.Process(target=minhash_multi_process, args=(sub_sets_files[i], entries_q, param_dic, i)) |
no test coverage detected