Python中如何避免向不同类实例重复传递Spark session等公共资源

以下是三种不同适用场景的实现方案，均无需显式向每个类传递SparkSession参数：

方案1：利用SparkSession内置单例特性 #

PySpark的SparkSession.builder.getOrCreate()本身就是单例实现，同一进程内任意位置调用都会返回同一个已初始化的实例，不会重复创建资源，是改造成本最低的方案。

修改后etl.py代码：

# etl.py
from pyspark.sql import SparkSession

class Source:
    """Class to organize ETL specific for `Source` dataframe"""
    def etl(self):
        spark = SparkSession.builder.getOrCreate()
        return spark.read.table('db.source_table')

class OtherSource:
    """Class to organize ETL specific for `OtherSource` dataframe"""
    def etl(self):
        spark = SparkSession.builder.getOrCreate()
        return spark.read.table('db.other_source_table').drop_duplicates()

class JoinedSources:
    """Class to organize ETL for `JoinedSources` dataframe"""
    def etl(self):
        spark = SparkSession.builder.getOrCreate()
        source_processed = Source().etl()
        other_source_processed = OtherSource().etl()
        return source_processed.join(other_source_processed, on='key')

修改后main.py代码：

# main.py
from pyspark.sql import SparkSession
from etl import JoinedSources

# 全局只需要初始化一次
spark = SparkSession.builder.getOrCreate()
joined_sources_processed = JoinedSources().etl()

# 其他业务逻辑
...

spark.stop()

适用场景：只有SparkSession需要共享、不想修改现有代码结构的场景

方案2：基类封装共享资源 #

如果后续还需要共享其他配置、资源，可以把公共逻辑抽象到父类中，所有ETL类继承父类即可统一获取共享资源，扩展性更强。

修改后etl.py代码：

# etl.py
from pyspark.sql import SparkSession

# 公共父类，所有ETL类继承该类即可获取共享资源
class EtlBase:
    _shared_spark = None
    @classmethod
    def init_spark(cls, spark):
        cls._shared_spark = spark

class Source(EtlBase):
    def etl(self):
        return self._shared_spark.read.table('db.source_table')

class OtherSource(EtlBase):
    def etl(self):
        return self._shared_spark.read.table('db.other_source_table').drop_duplicates()

class JoinedSources(EtlBase):
    def etl(self):
        source_processed = Source().etl()
        other_source_processed = OtherSource().etl()
        return source_processed.join(other_source_processed, on='key')

修改后main.py代码：

# main.py
from pyspark.sql import SparkSession
from etl import EtlBase, JoinedSources

spark = SparkSession.builder.getOrCreate()
# 全局只需要给父类赋值一次，所有子类都可共享该实例
EtlBase.init_spark(spark)

joined_sources_processed = JoinedSources().etl()
...

spark.stop()

适用场景：除了Spark还有其他共享配置/资源、需要统一管控公共逻辑的场景

方案3：依赖注入容器 #

如果是中大型项目，有非常多需要共享的资源、配置，可以用简单的依赖注入容器统一管理所有资源，全局统一注册、取用，资源生命周期管控更清晰。

示例实现：

# 通用依赖注入容器实现
class DIContainer:
    _dependencies = {}
    @classmethod
    def register(cls, dep_name: str, instance):
        cls._dependencies[dep_name] = instance
    @classmethod
    def get(cls, dep_name: str):
        return cls._dependencies.get(dep_name)

# main.py中全局注册资源
spark = SparkSession.builder.getOrCreate()
DIContainer.register("spark_session", spark)

# 任意ETL类中直接取用即可
class Source:
    def etl(self):
        spark = DIContainer.get("spark_session")
        return spark.read.table('db.source_table')

适用场景：中大型项目、依赖资源多、调用层级深的场景

注意事项 #

• 所有方案都需保证SparkSession生命周期可控，程序结束前主动调用spark.stop()释放资源，避免资源泄漏
• SparkSession不支持跨进程传递，若涉及多进程场景需要为每个进程单独初始化实例

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