如何在PySpark中访问Executor的Java Runtime变量（如maxMemory()）

你尝试的两种方法都因Spark的Driver/Worker架构限制失败，具体原因和解决方法如下：

失败原因分析 #

方法一错误原因 #

你在map算子中直接引用sc._gateway，但sc（SparkContext）是Driver端专属对象，无法在Worker端的计算逻辑中调用，因此触发SPARK-5063相关错误。
代码：

# 创建RDD
l = sc.range(100)
# 错误代码
l.map(lambda x: sc._gateway.jvm.java.lang.Runtime.getRuntime().maxMemory()).collect()

错误提示：

Exception: It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transformation. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063.
翻译：异常：你正尝试从广播变量、Action或Transformation中引用SparkContext。SparkContext仅能在Driver端使用，不能在Worker端运行的代码中引用。更多信息请查看SPARK-5063。

方法二错误原因 #

你在Driver端提前获取了Runtime对象，这个对象包含_thread.RLock这类无法序列化的锁资源，Spark需要将算子中的变量序列化后传到Worker端，因此触发序列化错误。
代码：

func = sc._gateway.jvm.java.lang.Runtime.getRuntime()
# 错误代码
l.map(lambda x: func.maxMemory()).collect()

错误提示：

TypeError: cannot pickle '_thread.RLock' object
翻译：类型错误：无法序列化'_thread.RLock'对象

正确解决方案 #

要在Worker端获取Executor JVM的maxMemory，需要在Worker本地的计算逻辑中直接创建JavaGateway连接到当前Executor的JVM，而非从Driver传递对象或引用SparkContext。推荐使用mapPartitions（每个分区仅执行一次，避免重复调用）：

from py4j.java_gateway import JavaGateway

def get_executor_max_memory(iterator):
    # 在Worker端创建JavaGateway，连接到当前Executor的JVM
    gateway = JavaGateway()
    runtime = gateway.jvm.java.lang.Runtime.getRuntime()
    # 返回当前Executor的maxMemory值，每个分区返回一次即可
    yield runtime.maxMemory()

# 创建测试RDD
rdd = sc.range(100)
# 执行计算并去重（同一个Executor上的多个分区结果一致）
max_memory_list = rdd.mapPartitions(get_executor_max_memory).distinct().collect()

# 输出每个Executor的maxMemory值
for mem in max_memory_list:
    print(f"Executor JVM maxMemory: {mem} bytes")

补充说明 #

• 使用mapPartitions而非map，可减少每个Executor上的调用次数，提升效率；
• 调用distinct()是因为同一个Executor上的多个分区会返回相同结果，去重后能看到所有Executor的唯一值；
• 每个Worker进程对应一个Executor JVM，因此该方法能准确获取每个Executor的内存参数。

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