在生产环境中运行PySpark作业时，仅依赖Spark内置JVM内存参数（如 spark.driver.memory）往往无法实现对Python层内存使用的有效约束。尤其在Standalone模式下，PySpark会启动多层级进程：JVM侧的Driver/Executor与Python侧的Py4J网关、Worker子进程（pyspark.daemon 和 python worker），二者内存空间相互独立。若不对Python进程驻留内存（RSS）施加硬性限制，易导致系统OOM Killer介入或资源争抢，影响集群稳定性。 本文基于 PySpark 3.1.2 + Linux cgroups v1 环境，提出一种绕过JVM虚拟内存限制、直接管控物理内存占用的实践方案。

PySpark内存结构再梳理

PySpark任务实际由四类关键进程构成，需分别设限：

• JVM Driver（spark-submit）：承载SparkContext、DAG调度逻辑，受 spark.driver.memory 控制；
• Python Driver（python主进程）：执行用户Python脚本（如sc.parallelize()、df.collect()），其内存不受Spark配置影响；
• JVM Executor（CoarseGrainedExecutorBackend）：执行Shuffle、序列化等JVM任务，由 spark.executor.memory 限定；
• Python Worker（python -m pyspark.worker）：执行map、filter等UDF，其RSS内存增长不可控，是资源泄漏高发区。

其中，spark.executor.pyspark.memory 虽可触发 resource.setrlimit(RLIMIT_AS)，但该机制限制的是虚拟地址空间（VIRT），而非实际物理内存（RSS）。实测表明：当设置为 2g 时，VIRT可能达2GB而RSS仅300MB，造成内存配额严重闲置。

cgroups：精准限制RSS内存的核心手段

Linux cgroups v1 的 memory 子系统提供 memory.limit_in_bytes 接口，可对进程组的RSS+Cache总和实施硬性上限，并在超限时触发OOM Killer终止违规进程——这正是我们需要的物理内存围栏。

分层控制策略

为解耦管理，建立三级cgroup目录结构：

/sys/fs/cgroup/memory/pyspark/
├── driver/      # 约束 Python Driver 进程（含 spark-submit JVM）
└── executor/    # 约束所有 Python Worker 子进程

操作步骤如下：

1. 创建分组：

   sudo mkdir -p /sys/fs/cgroup/memory/pyspark/{driver,executor}

2. 设定内存上限（以16GB物理内存机器为例）：

   # Driver组：预留4GB（含JVM+Python Driver）
   sudo sh -c "echo $((4 * 1024 * 1024 * 1024)) > /sys/fs/cgroup/memory/pyspark/driver/memory.limit_in_bytes"
   
   # Executor组：预留10GB（覆盖全部Worker进程）
   sudo sh -c "echo $((10 * 1024 * 1024 * 1024)) > /sys/fs/cgroup/memory/pyspark/executor/memory.limit_in_bytes"

3. 注入进程PID：
   • 启动PySpark前，先获取 spark-submit 主进程PID（如 160224），加入driver组：

   echo 160224 | sudo tee /sys/fs/cgroup/memory/pyspark/driver/cgroup.procs

   • Executor Worker进程由Spark动态派生，需在作业启动后捕获其PID。可通过以下命令实时监控并注入：

   # 监听新生成的 python worker 进程（过滤 pyspark.worker）
   sudo ps aux --forest | grep "pyspark\.worker" | grep -v grep | awk '{print $2}' | \
     xargs -r -I{} sh -c 'echo {} | sudo tee -a /sys/fs/cgroup/memory/pyspark/executor/cgroup.procs'

自动化注入脚本示例

为避免手动追踪PID，可编写轻量级守护脚本，在Executor启动后自动归集：

#!/usr/bin/env python3
import os
import time
import subprocess

EXECUTOR_CGROUP = "/sys/fs/cgroup/memory/pyspark/executor/cgroup.procs"
WORKER_PATTERN = "pyspark.worker"

def get_worker_pids():
    try:
        result = subprocess.run(
            ["pgrep", "-f", WORKER_PATTERN],
            capture_output=True, text=True, check=True
        )
        return [int(pid.strip()) for pid in result.stdout.splitlines() if pid.strip()]
    except subprocess.CalledProcessError:
        return []

def inject_to_cgroup(pids):
    with open(EXECUTOR_CGROUP, "a") as f:
        for pid in pids:
            try:
                f.write(f"{pid}\n")
                print(f"Injected PID {pid}")
            except OSError as e:
                if "No such process" not in str(e):
                    print(f"Failed to inject {pid}: {e}")

if __name__ == "__main__":
    seen = set()
    while True:
        current = set(get_worker_pids())
        new_pids = current - seen
        if new_pids:
            inject_to_cgroup(list(new_pids))
            seen.update(new_pids)
        time.sleep(2)

将此脚本作为后台服务运行，即可持续保障所有Worker进程落入cgroup管控。

验证与调试

检查当前组内存使用：

cat /sys/fs/cgroup/memory/pyspark/driver/memory.usage_in_bytes
cat /sys/fs/cgroup/memory/pyspark/executor/memory.usage_in_bytes

查看被OOM Killer终结的进程记录：

dmesg | grep -i "killed process"

若发现Worker频繁被kill，可调高 memory.limit_in_bytes 值，或检查Python UDF是否存在内存泄漏（如未释放大对象引用、全局缓存无界增长等）。

注意事项

• cgroups v1 在容器化环境（Docker/K8s）中需额外启用 --memory 参数，且需确保内核支持 CONFIG_MEMCG；
• 同一进程不能同时属于多个memory cgroup，迁移PID时需先从原组移除；
• Spark Standalone的Executor重启后会产生新PID，必须重新注入cgroup，否则失去管控；
• 该方案不替代JVM参数，应与 spark.driver.memory、spark.executor.memory 协同配置，形成JVM堆+Python RSS双维度防护。