Этот вызов multiprocess() является частью гораздо более крупного сценария, который я пишу в Spyder. Причина, по которой я делаю этот отдельный read_parquet, заключается в том, что я прочитал, что multiprocess() не может видеть переменные, которые у меня есть в моей среде Spyder, поэтому я должен принять их снова.
Размер материала и фрагмента 1 предназначен только для тестирования; Я удалю все это, как только код начнет работать с меньшими наборами данных.
Основные проблемы на данный момент:
- Код работает в Spyder, но продолжает работать, и я не вижу активности в диспетчере задач. Я вижу, что запускаются новые процессы, но загрузка их процессора остается равной 0, а объем используемой памяти не меняется. Я смотрел это около 30 минут.
- Я почти уверен, что мой код где-то неверен. Я использовал различные стратегии, чтобы заставить что-то работать, но все они потерпели неудачу. Я пробовал использовать асинхронный запуск и очереди для сбора результатов. Все безрезультатно.
Обновление 17 февраля 2026 г.: использовал Copilot, чтобы получить несколько советов по кодированию; все еще не работает; файлы журналов не созданы; не вижу увеличения использования процессора в диспетчере задач, как я ожидал. Я вижу, что процессор открыт и память используется, но процессор неактивен.
import pyarrow.parquet as pq
from rapidfuzz import process, fuzz
from rapidfuzz.utils import default_process
import pandas as pd
import duckdb
import os
import numpy as np
import multiprocessing as mp
import logging, os, traceback
def fuzzy_match_mike(security_4c, df_hold_choices): # add back in result_queue as an arg once verified
logging.basicConfig(
filename = f'worker_{os.getpid()}.log',
level = logging.DEBUG,
format = '%(asctime)s %(levelname)s %(messages)s'
)
try:
best_match = []
best_match2 = []
best_match3 = []
score = []
score2 = []
score3 = []
index = []
index2 = []
index3 = []
result_store_prime = []
for i in range(security_4c.shape[0]):
print(f"This is the df for this child process: {security_4c}")
df_temp = df_hold_choices.copy(deep=True)
bm, sc, idx = process.extractOne(
security_4c['4c_primary_name2'].tolist(),
df_temp,
scorer=fuzz.WRatio,
score_cutoff = 80,
# since we preprocessed the data, no need to have rapidfuzz process the records
processor=None
)
df_temp[idx] = np.nan
bm2, sc2, idx2 = process.extractOne(
security_4c['4c_primary_name2'].tolist(),
df_temp,
scorer=fuzz.WRatio,
score_cutoff = 80,
processor=None
)
df_temp[idx2] = np.nan
bm3, sc3, idx3 = process.extractOne(
security_4c['4c_primary_name2'].tolist(),
df_temp,
scorer=fuzz.WRatio,
score_cutoff = 80,
processor=None
)
best_match.append(bm)
score.append(sc)
index.append(idx)
best_match2.append(bm2)
score2.append(sc2)
index2.append(idx2)
best_match3.append(bm3)
score3.append(sc3)
index3.append(idx3)
result = [i,best_match,
score,
index,
best_match2,
score2,
index2,
best_match3,
score3,
index3]
result_store_prime.append(result)
except Exception:
logging.exception("Unhandled exception in fuzzy_match_mike")
traceback.print_exc()
raise
return result_store_prime
# ________________________END PRIMARY FUNCTION DECLARATION_____________________
security_4c['4c_primary_name'] = security_4c['4c_primary_name'].astype(str).fillna("")
security_4c['4c_primary_name2'] = [default_process(i) for i in security_4c['4c_primary_name'].tolist()]
df_hold['sba_data_primary_name'] = df_hold['sba_data_primary_name'].astype(str).fillna("")
df_hold['sba_data_primary_name2'] = [default_process(i) for i in df_hold['sba_data_primary_name'].tolist()]
df_hold_choices = df_hold['sba_data_primary_name2']
stuff = security_4c.iloc[12:16,]
os.chdir(r"\\sharedDrive\current_projects\ALR\Securities")
df_hold.to_parquet("df_hold.parquet", engine = 'pyarrow', compression = 'snappy')
security_4c.to_parquet("security_4c.parquet", engine = 'pyarrow', compression = 'snappy')
if __name__=='__main__':
# with statement allows pools to close and join on their own
os.chdir(r"\\sharedDrive\current_projects\ALR\Securities")
df_hold = pd.read_parquet("df_hold.parquet",
thrift_string_size_limit = 1000000000,
thrift_container_size_limit = 1000000000)
security_4c = pd.read_parquet("security_4c.parquet",
thrift_string_size_limit = 1000000000,
thrift_container_size_limit = 1000000000)
df_hold_choices = df_hold['sba_data_primary_name2']
with mp.Pool(processes = mp.cpu_count()) as pool:
stuff = security_4c.iloc[12:16,]
chunk_size = 1
# Starmap needs to be passed a tuple; each of which needs to be a
# standalone packet that the function accepts. Remember, the function
# inputs are (4c_data, sba data to compare to)
list_of_dfs = [stuff[i:i+chunk_size] for i in range(0, len(stuff), chunk_size)] #Update to all of 4c once working
list_of_dfs = [(df_un, df_hold_choices) for df_un in list_of_dfs]
results = pool.starmap(fuzzy_match_mike, list_of_dfs)
final_result = [item for sublist in results for item in sublist]
Подробнее здесь: https://stackoverflow.com/questions/798 ... sk-manager
