使用 mpi4py 后在 Pandas 中聚合结果

Question

最后，这是我关于 StackOF 的第一个问题：

作为 uni 的一个项目，我正在尝试从头开始为 KMeans 编写代码，然后使用 mpi4py 并行运行具有随机起始中心的不同重复。

代码如下：

#!/usr/bin/env python
# coding: utf-8

# In[3]:
import numpy as np
import matplotlib.pyplot as plt
import pandas as pd
from mpi4py import MPI
# import statistics as stat

comm=MPI.COMM_WORLD
rank = comm.Get_rank()
size = comm.Get_size()
print('no of processors is', size)
print('this is the process #', rank)
df = pd.read_csv('data.dat',
                   sep='   ',
                   header=None,
                   index_col=0, engine='python' )

n_mus = [1, 2, 4, 12]  # 100]#, 1000]
cost_k = []
k_vals = range(1, 5, 2)
# k_vals = range(1, 30, 6)

for orig_n_mu in n_mus:
    n_mu = orig_n_mu//size
    if rank in range(orig_n_mu%size):
        n_mu += 1
    for k in k_vals:
        cost_n = []
        for n in range(1, n_mu + 1):
            np.random.seed(n * k + k)
            kx = np.random.uniform(df[1].min(), df[1].max(), k)
            np.random.seed(n * k + k + 1)
            ky = np.random.uniform(df[2].min(), df[2].max(), k)
            manh = pd.DataFrame()
            for c in range(k):
                manh[c] = abs(df[1] - kx[c]) + abs(df[2] - ky[c])
            df['center'] = manh.idxmin(axis='columns')
            kx = df.groupby('center').mean()[1]
            ky = df.groupby('center').mean()[2]
            if df.center.unique().shape[0] != k:
                print('not all centers took up clusters at the number', n,
                      'repetition')
                print('the current number of clusters is:',
                      df.center.unique().shape[0], 'instead of', k)
            diff = 10
            while diff > 1e-4:
                cost = manh.min(axis=1).mean()
                for c in df.center.unique():
                    manh[c] = abs(df[1] - kx[c]) + abs(df[2] - ky[c])
                df['center'] = manh.idxmin(axis='columns')
                kx = df.groupby('center').mean()[1]
                ky = df.groupby('center').mean()[2]
                new_cost = manh.min(axis=1).mean()
                diff = cost - new_cost
            cost_n.append(new_cost)
        cost_k.append([k, rank, n_mu, orig_n_mu, cost_n])
print('process #', rank, 'is done here')
all_cost = comm.gather(cost_k, root = 0)
if (rank == 0):
    print('check point #1')
    all_cost = np.reshape(all_cost, newshape=(-1,len(cost_k[0])))
    print('the shape of all cost is', all_cost.shape)
    res = pd.DataFrame(all_cost, columns=['k_val', 'rank', 'n_mu', 'orig_n_mu','cost_res'])
    noruns = (res.n_mu == 0)
    res = res[~noruns].copy()
    res.reset_index(inplace=True, drop=True)
    print('check point #2')
    cost_funcs = pd.DataFrame(res.cost_res.to_list())
    print('check point #3')
    km_df = pd.merge(res, cost_funcs, how='outer',left_index=True, right_index=True)
    print('check point #4')
    km_df.drop(columns='cost_res', inplace = True)

    km_df['avg_final_cost'] = cost_funcs.apply(np.nanmean, axis =1)
    km_df['std_final_cost'] = cost_funcs.apply(np.nanstd, axis =1)
    km_df['min_final_cost'] = cost_funcs.apply(min, axis =1)
    km_df['max_final_cost'] = cost_funcs.apply(max, axis =1)

    km_df.to_csv('km_df_test_para.csv')

# km_df

生成的 csv 如下所示： sample csv screenshot

这里的 n 是每个核心上的运行次数，orig_n 是我应该进行分析、记录时间、检查标准、平均值等的运行总数。列 0、1、2、...是每次运行的结果，列名是单个核心上的运行次数。

现在我需要将所有这些运行按 n_orig 分组。但是不知道如何告诉熊猫将所有具有相同 n_orig 和 k 的值放在同一行中。如您所知，我对 mpi 也很陌生，不知道如何收集我的数据。 Gather 和 Gatherv 命令不断删除错误“0”。

如果您能提供任何帮助，我将不胜感激：)

Answer 1

我不完全理解你的意思，但也许按n_orig 和k_val 对行进行排序就足够了？

km_df.sort_values(by=['n_orig', 'k_val'], inplace=True)

或者您可以将km_df 拆分为更小的常量n_orig 和k_val 数据帧：

from itertools import product

groups = []
for n_orig, k_val in product(km_df['n_orig'].unique(), km_df['k_val'].unique()):
    sel = (km_df['n_orig'] == n_orig) & (km_df['k_val'] == k_val)
    groups.append(km_df[sel])