链接到Pandas DataFrame的转换

如何解决链接到Pandas DataFrame的转换

我有一个非常宽的DataFrame（〜10K列），可以将其用作模型的训练数据。从不同来源的其他几个表进行构建大约需要30分钟。我的问题是针对此表的预处理和构建的。我正在以不同的清洁度和完成状态组装同一数据的不同来源。这种预处理是在几个转换阶段使用难以理解和维护的代码完成的。我想问一下是否有更好的方法或工具来构造这种转换，从而使代码更易读，简洁和易于维护。

现在，该代码看起来像是此代码段的精简版本：

# Data sources

data1 = {
    'id': range(10),'feat_A_source_1_type_0': np.random.rand(10),'feat_A_source_1_type_3': np.random.rand(10),'feat_B_source_1_type_3': np.random.rand(10),'feat_C_source_1_type_0': np.random.rand(10),'feat_D_source_1_type_1': np.random.rand(10)
}
df_A = pd.DataFrame(data1,columns=list(data1.keys()))
df_A.set_index('id',inplace=True)

data2 = {
    'id': range(10),'feat_A_source_2_type_3': np.random.rand(10),'feat_B_source_2_type_0': np.random.rand(10),'feat_B_source_2_type_3': np.random.rand(10),'feat_D_source_2_type_2': np.random.rand(10),'feat_D_source_2_type_3': np.random.rand(10)
}
df_B = pd.DataFrame(data2,columns=list(data2.keys()))
df_B.set_index('id',inplace=True)

# Preprocessing specs

final_features = ['feat_A','feat_B','feat_C','feat_D']
preference_sources = {
    'feat_A': ['source_2','source_1'],'feat_B': ['source_3'],'feat_C': ['source_1'],'feat_D': ['source_1','source_2'],}
feature_specs = {
    'feat_t012':   (np.median,['feat_A_type_1','feat_B_type_0','feat_A_type_2']),'feat_f02':    (np.sum,['feat_D_type_3','feat_A_type_1','feat_C_type_2']),'feat_B2':     (np.mean,['feat_B_type_3','feat_B_type_0']),'feat_ZZ_std': (np.std,['feat_A_type_3','feat_C_type_1','feat_C':      (np.sum,['feat_C_type_0']),'feat_D_m':    (np.mean,['feat_D_type_2','feat_D_type_3','feat_D_type_0']),}

# Actual preprocessing (after data cleaning for each source)

merged_df = pd.concat([df_A,df_B],axis=1)

## Applying preference
pref_data = {}
for feat,prefs in preference_sources.items():
    data = []
    for pref in prefs:
        cols = {c: c.replace(pref,'').replace('__','_') for c in merged_df if f'{feat}_{pref}' in c}
        m = merged_df[cols.keys()].copy()
        m.rename(columns=cols,inplace=True)
        data.append(m)
    p = data[0]
    for n in data[1:]:
        n = n[set(n.columns) - set(p.columns)]
        p = pd.concat([p,n],axis=1)
    pref_data[feat] = p
pref_data = pd.concat(pref_data.values(),axis=1)

## Applying specs
## In reality the specs contain several layers,so multiple passes are needed to get the final features

specs_data = {}
for final_feat,(func,feats) in feature_specs.items():
    data = pref_data[[f for f in feats if f in pref_data.columns]]
    specs_data[final_feat] = data.apply(func=func,axis=1)
final_df = pd.DataFrame(specs_data)

这是一个非常简化的版本。实际上，这里使用了更复杂的功能，有时会使用两组具有不同功能名称的输入（已通过首选项选择）。等同于：

df['feat_M'] = np.median(df[[c for c in df.columns if 'feat_D' in c]]) / np.sum(df[[c for c in df.columns if 'feat_S_source_2' in c]])

我认为，像Spark或Dask这样的工具（尤其是流媒体）很好用是一项任务。恐怕它们需要一些设置，并且代码需要进行修改才能使用它们，因此您只能使用该框架。此外，如果您没有设置，就不能只在本地运行测试，因此代码共享很困难。

我发现pyjanitor看起来不错，但是没有我需要的灵活性。此外，数据集随着时间的推移而增加，几乎无法容纳在内存中。在一段时间内，我将不得不面对一个事实，那就是我需要使用k8s或Spark，因此，如果有一个也可以朝这个方向扩展的工具，那就太好了。

如果有人有经验或知道一种获得更佳，更简单的代码以达到相似结果的方法，那就太好了。