PySpark调用foreachPartition出现无法序列化_thread.RLock报错如何解决

该报错由Spark闭包序列化失败导致。你定义的func1是SampleClass类__init__方法内的嵌套函数，它隐式携带了所属类实例的引用，而类实例中持有spark会话对象，其中包含不可序列化的_thread.RLock锁资源，当Spark尝试把func1序列化分发到Executor节点执行时就会触发该错误。

你可以通过以下任意一种方式调整代码结构解决问题：

方案1：将分区处理函数抽为独立函数 #

把func1从类内部挪到全局作用域，避免携带类实例引用，是最稳妥的解决方式：

# 独立的分区处理函数，不绑定任何类实例
def func1(partition):
    print(partition.getPartitionId)
    print(partition.count())
    # 其他分区处理逻辑

class SampleClass:
    def __init__(self, spark):
        df = (spark.read.table("sampletable").repartition(100))
        df.rdd.foreachPartition(func1)

def main():
    from pyspark.sql import SparkSession
    spark = SparkSession.builder.appName("sample_app").getOrCreate()
    # 注意原代码实例化SampleClass时漏传了spark参数，这里需要补上
    s = SampleClass(spark)

if __name__ == "__main__":
    main()

方案2：将分区处理函数设为类静态方法 #

如果要保留函数在类内的封装结构，可以将其声明为静态方法，避免隐式携带self实例引用：

class SampleClass:
    def __init__(self, spark):
        df = (spark.read.table("sampletable").repartition(100))
        df.rdd.foreachPartition(SampleClass.func1)
    
    @staticmethod
    def func1(partition):
        print(partition.getPartitionId)
        print(partition.count())
        # 其他分区处理逻辑

def main():
    from pyspark.sql import SparkSession
    spark = SparkSession.builder.appName("sample_app").getOrCreate()
    s = SampleClass(spark)

if __name__ == "__main__":
    main()

• 不要在需要序列化分发到Executor的函数里引用SparkSession、SparkContext这类仅存在于Driver端的不可序列化对象
• 如果需要向分区处理函数传递参数，仅传递字符串、数值、列表、字典这类可序列化的基础类型数据，不要传递自定义类实例、连接句柄、锁等不可序列化资源
• 原代码实例化SampleClass时未传入要求的spark参数，运行时会额外触发参数缺失报错，需要补充

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