基于pymfe的大数据集元特征分析内存优化：分块计算与合并

当使用pymfe处理大规模数据集时，内存占用过高导致进程被终止，可以通过拆分数据集为小批次计算，再合并统计结果的方式解决。以下是具体实现步骤和代码：

核心思路 #

将大数据集拆分为多个小子集，分别计算每个子集的元特征统计量（min/max/mean/sd），再根据统计量类型采用对应方法合并：

• min：取所有批次的最小值
• max：取所有批次的最大值
• mean：按各批次样本数加权计算平均值
• sd：用合并方差公式计算整体标准差

代码实现 #

1. 导入依赖并生成模拟数据集 #

import numpy as np
from sklearn.datasets import make_classification
from pymfe.mfe import MFE

# 模拟用户的大规模数据集（32690样本，80特征）
X, y = make_classification(n_samples=32690, n_features=80, n_informative=60, n_classes=5, random_state=42)

2. 拆分数据集为小批次 #

def split_dataset(X, y, batch_size=2000):
    """将数据集拆分为指定大小的小批次"""
    n_samples = X.shape[0]
    batches = []
    for i in range(0, n_samples, batch_size):
        end_idx = min(i + batch_size, n_samples)
        batches.append((X[i:end_idx], y[i:end_idx]))
    return batches

# 拆分批次，可根据内存调整batch_size（比如1000/5000）
batches = split_dataset(X, y, batch_size=2000)

3. 批量计算每个子集的元特征 #

# 存储各批次的统计结果：键为统计类型，值为对应结果+样本数（按需）
batch_results = {
    "min": [],
    "max": [],
    "mean": [],
    "sd": [],
}

for batch_X, batch_y in batches:
    # 初始化MFE提取器，指定目标特征和统计量
    extractor = MFE(features=["t1"], groups=["complexity"], summary=["min", "max", "mean", "sd"])
    extractor.fit(batch_X, batch_y)
    _, feat_vals = extractor.extract()
    
    # 提取当前批次的统计值和样本数
    batch_min, batch_max, batch_mean, batch_sd = feat_vals
    batch_size = batch_X.shape[0]
    
    # 存入结果字典
    batch_results["min"].append(batch_min)
    batch_results["max"].append(batch_max)
    batch_results["mean"].append((batch_mean, batch_size))
    batch_results["sd"].append((batch_sd, batch_mean, batch_size))

4. 合并各批次的统计结果 #

final_results = {}

# 合并min：取所有批次最小值
final_results["t1_min"] = min(batch_results["min"])

# 合并max：取所有批次最大值
final_results["t1_max"] = max(batch_results["max"])

# 合并mean：加权平均（按样本数权重）
total_samples = sum(size for _, size in batch_results["mean"])
weighted_sum = sum(mean * size for mean, size in batch_results["mean"])
final_results["t1_mean"] = weighted_sum / total_samples

# 合并sd：使用合并方差公式计算整体标准差
sum_var = 0.0
sum_mean_diff = 0.0
mean_total = final_results["t1_mean"]

for sd, mean, size in batch_results["sd"]:
    sum_var += (size - 1) * (sd ** 2)
    sum_mean_diff += size * (mean - mean_total) ** 2

var_total = (sum_var + sum_mean_diff) / (total_samples - 1)
final_results["t1_sd"] = np.sqrt(var_total)

# 输出最终结果
print("合并后的元特征结果：")
print(f"t1_min: {final_results['t1_min']:.4f}")
print(f"t1_max: {final_results['t1_max']:.4f}")
print(f"t1_mean: {final_results['t1_mean']:.4f}")
print(f"t1_sd: {final_results['t1_sd']:.4f}")

注意事项 #

• 批次大小batch_size可根据内存情况灵活调整：内存紧张时调小（如1000），内存充足时调大（如5000）。
• 如果需要提取多个元特征，只需修改features参数，并在批量计算时对应存储每个特征的结果即可。
• 该方法适用于大多数数值型统计量的合并，对于特殊元特征需根据其定义调整合并逻辑。

内容的提问来源于stack exchange，提问作者arilwan