AWS EMR Zeppelin运行Spark代码报Python进程异常退出如何解决

根因说明 #

你使用的pre_stage_1.toPandas()会将分布式存储的百万行Spark DataFrame全量拉取到Zeppelin所在的Driver节点Python进程内存中，搭配单节点运行的sklearn KNNImputer计算时，内存开销超过进程阈值触发OOM，被系统强制终止，因此抛出Python进程异常退出报错。

修复方案 #

方案1：调优Driver端内存配置（临时应急） #

仅适合临时跑小批量任务，无法解决单节点算力瓶颈，数据量增长后仍会触发报错：

• 进入EMR控制台的Zeppelin配置界面，找到Spark Interpreter的spark.driver.memory参数，将默认值调大到8G及以上
• 调整完成后重启Spark Interpreter，重新运行代码即可

方案2：用Spark MLlib分布式KNNImputer替换（最优选择） #

全程用集群分布式算力计算，不需要把数据拉到单节点，无内存瓶颈，适配百万到亿级数据集，示例代码如下：

%spark.pyspark
from pyspark.ml.feature import KNNImputer, VectorAssembler
from pyspark.ml.functions import vector_to_array

# 1. 将要填充的数值列合并为特征向量列，替换下方列表为你实际要填充的列名
fill_cols = ["col1", "col2", "col3"]
assembler = VectorAssembler(inputCols=fill_cols, outputCol="features", handleInvalid="keep")
df_with_features = assembler.transform(pre_stage_1)

# 2. 运行分布式KNN填充
knn_imputer = KNNImputer(inputCol="features", outputCol="imputed_features", k=5)
imputed_df = knn_imputer.fit(df_with_features).transform(df_with_features)

# 3. 将填充后的向量列拆回原有列结构
imputed_df = imputed_df.withColumn("imputed_arr", vector_to_array("imputed_features"))
for idx, col_name in enumerate(fill_cols):
    imputed_df = imputed_df.withColumn(col_name, imputed_df.imputed_arr[idx])

# 4. 得到最终填充后的数据集
pre_stage_1 = imputed_df.select(pre_stage_1.columns)

方案3：分区分批处理（仅适合允许按分区填充的场景） #

如果必须使用sklearn的KNNImputer实现，可按Spark分区分批拉取计算，避免全量数据压到单节点：

%spark.pyspark
import pandas as pd
from sklearn.impute import KNNImputer

def process_partition(partition_iter):
    # 单次仅加载单个分区的数据到内存处理
    partition_pdf = pd.DataFrame(list(partition_iter), columns=pre_stage_1.columns)
    knn_imputer = KNNImputer()
    partition_pdf.iloc[:, :] = knn_imputer.fit_transform(partition_pdf)
    for row in partition_pdf.itertuples(index=False):
        yield tuple(row)

# 分布式运行分区级填充逻辑
pre_stage_1 = pre_stage_1.rdd.mapPartitions(process_partition).toDF(pre_stage_1.columns)

注意：该方案的KNN填充仅参考同分区内的样本特征，和全量数据集计算得到的填充结果存在差异，需要业务侧接受该差异再使用。

如果调整后仍报错，可查看EMR Yarn ResourceManager日志或者Zeppelin本地日志，确认内存不足的节点是Driver还是Executor，对应调整相关内存参数即可。

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