PySpark报'NoneType' object has no attribute '_jvm'错误：Spark2.2统计DataFrame分区元素数

看起来你已经走在正确的路上了——自定义的count_elements函数逻辑没问题，问题出在代码的完整性和一些PySpark的细节上。咱们先把错误解决，再实现完整的分区统计功能。

修复后的可运行代码 #

from pyspark.sql.functions import *
from pyspark.sql import SparkSession

def count_elements(splitIndex, iterator):
    n = sum(1 for _ in iterator)
    yield (splitIndex, n)

if __name__ == "__main__":
    # 初始化SparkSession，确保在driver进程中正确实例化
    spark = SparkSession.builder.appName("PartitionCount").getOrCreate()
    
    num_parts = 3
    # 读取JSON数据并重新分区
    df = spark.read.json("/tmp/tmp/gon_s.json").repartition(num_parts)
    
    print(f"df has {df.rdd.getNumPartitions()} partitions.")
    
    # 用mapPartitionsWithIndex绑定自定义函数，收集结果到driver端
    partition_counts = df.rdd.mapPartitionsWithIndex(count_elements).collect()
    
    print("Elements across partitions:")
    for part_idx, count in partition_counts:
        print(f"Partition {part_idx}: {count} elements")
    
    # 显式关闭SparkSession
    spark.stop()

错误原因拆解 #

1. 未完成的代码引发None引用：你原来的最后一行print("Elements across partitions is:" + str(df....没有写完，大概率是想调用mapPartitionsWithIndex但没完成，导致代码尝试对一个None对象进行操作——而Spark内部很多核心操作依赖_jvm对象，当引用无效时就会抛出'NoneType' object has no attribute '_jvm'错误。
2. 缺失进程防护导致初始化异常：在PySpark本地模式下，没有if __name__ == "__main__":的话，worker进程会重复执行脚本，可能导致SparkSession初始化失败，进而出现_jvm相关的None错误。

核心逻辑说明 #

• mapPartitionsWithIndex：这个方法会把每个分区的索引和分区内的元素迭代器传给你的count_elements函数，函数通过遍历迭代器统计元素数量，返回分区索引与对应数量的元组。
• collect()：将分布式集群上的分区统计结果拉取到driver进程，这样才能直接打印输出。
• 显式关闭SparkSession：虽然程序结束后Spark会自动关闭，但显式调用spark.stop()是更规范的做法，避免资源泄漏。

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