我有以下的函数,它会随机打乱数据框中一列的值,并对包括被随机打乱列的整个数据框使用RandomForestClassifier
来获取准确率分数。
我想同时对数据框的每一列运行这个函数,因为数据框非常大,包含500k行和1k列。关键是每次只随机打乱一列。
然而,我很困惑为什么ProcessPoolExecutor
比ThreadPoolExecutor
慢得多。我原以为ThreadPoolExecutor
只适用于I/O任务更快。在这种情况下,它不涉及读取或写入任何文件。
或者我做错了什么?有没有更高效或更好的方法来优化这段代码,以实现并发执行和提高速度?
def randomShuffle(colname, X, y, fit):
out = {'col_name': colname}
X_= X.copy(deep = True)
np.random.shuffle(X_[colname].values) # permutation of a single column
pred = fit.predict(X_)
out['scr'] = accuracy_score(y, pred)
return out
def runConcurrent(classifier, X,y):
skf = KFold(n_splits=5, shuffle = False)
acc_scr0, acc_scr1 = pd.Series(), pd.DataFrame(columns = X.columns)
# split data to training and validation
for i, (train_idx, val_idx) in enumerate(skf.split(X,y)):
X_train, y_train = X.iloc[train_idx,:], y.iloc[train_idx]
X_val, y_val = X.iloc[val_idx,:], y.iloc[val_idx]
fit = classifier.fit(X=X_train, y=y_train)
# accuracy score
pred = fit.predict(X_val)
acc_scr0.loc[i] = accuracy_score(y_val, pred)
# with concurrent.futures.ProcessPoolExecutor() as executor:
with concurrent.futures.ThreadPoolExecutor() as executor:
results = [executor.submit(randomShuffle, colname = j, X= X_val, y= y_val, fit = fit, labels = classifier.classes_) for j in X.columns]
for res in concurrent.futures.as_completed(results):
acc_scr1.loc[i, res.result()['col_name']] = res.result()['acc_scr']
return None