Spark独立集群内存分配失败及PySpark代码异常行为排查

我来帮你拆解这个奇怪现象背后的原因，其实这和Spark的懒执行机制密切相关，咱们一步步来分析：

问题核心：Spark的懒加载 vs 显式Action触发 #

首先要记住：Spark中DataFrame的转换操作（比如select、filter）都是懒执行的——它们不会立即计算结果，只会默默构建执行计划，直到遇到Action操作（比如count、collect、show，甚至print）才会真正触发计算。

你的代码里，当注释掉print(cur_df)时，所有计算逻辑（包括cur_df的生成、clrr_df的过滤统计、最后的join）会被Spark合并成一个超大的执行Stage。这个Stage里，单个Task需要同时处理大量中间数据，内存占用直接超过了分配阈值，于是就出现了Failed to allocate a page的内存告警。

而当你取消注释print(cur_df)时，print会触发一个Action操作，强制Spark先完成cur_df的计算并输出其Schema/部分数据。这相当于把整个计算流程拆成了两个独立的Job：

1. 第一个Job：计算cur_df并输出（由print触发）
2. 第二个Job：基于已经计算好的cur_df结果，执行后续的collect、count、join操作

拆分后的每个Job内存压力小了很多，Task能顺利分配到所需内存，自然就不会报错了。

你的代码里还有几个潜在风险点，顺便帮你指出来： #

• collect()滥用：cur_df.select(['cliNum']).collect()[0]['cliNum']会把cur_df中所有cliNum数据拉到Driver端，如果cur_df数据量很大，Driver很容易内存溢出。建议改成cur_df.select('cliNum').first()['cliNum']，只会拉取第一条数据，更安全。
• 重复计算：你两次执行clrr_df.filter('cliNum='+cli_num)，Spark会重复扫描clrr_df的数据，浪费资源。可以提前过滤并缓存：

  filtered_clrr = clrr_df.filter(clrr_df.cliNum == cli_num).cache()
  number_of_pols_lf = filtered_clrr.filter(filtered_clrr.clientRole == 'LF').count()
  number_of_pols_ow = filtered_clrr.filter(filtered_clrr.clientRole == 'OW').count()
  

• 硬编码字符串过滤：filter('cliNum='+cli_num)存在SQL注入风险（如果cliNum是字符串类型还会触发语法错误），建议用列表达式过滤，比如clrr_df.filter(clrr_df.cliNum == cli_num)。

替代print的更优雅解决方案 #

既然print的作用是提前触发cur_df的计算，那我们可以用更专业的缓存机制替代：

cur_df = cur_df.cache()  # 或者用persist(StorageLevel.MEMORY_AND_DISK)，根据数据量选择
cur_df.count()  # 触发缓存，相当于替代print的Action操作

这样既避免了不必要的打印输出，又能让Spark提前计算并缓存cur_df的结果，后续操作直接基于缓存数据，内存压力会小很多。

额外的集群配置建议 #

如果代码优化后还是有内存问题，可以调整Spark的内存参数：

• 增大Executor内存：--executor-memory 4g（根据你的集群资源调整）
• 调整Driver内存：--driver-memory 2g（如果Driver经常溢出的话）
• 调整内存分配比例：spark.memory.fraction和spark.memory.storageFraction，让执行内存有更多可用空间

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