SparkSession复用问题：Context类gc属性访问报错及优化咨询

我想在代码里通过getOrCreate()方法共享完整的SparkSession，但尝试访问Context类中实例化的gc时遇到如下错误：

AttributeError: type object 'Context' has no attribute 'gc'

我的目标是将SparkSession独立封装到类中，按需调用以避免逻辑耦合，尝试过继承、在RDSClient中直接调用Context类等方法，均未解决问题。

更新：我已实现一个可行方案，但想了解是否存在更优处理方式，当前代码如下：

import sys
import re
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
import awsglue
import pyspark
import boto3
import pymssql
import pandas

class Context():
    def __init__(self):
        self.sc = SparkContext()
        self.gc = GlueContext(self.sc.getOrCreate())
        self.spark = self.gc.spark_session
    
    @staticmethod
    def get_parameters(*args):
        return getResolvedOptions(sys.argv, ["dynamodb_metadata_table"])

class RDSClient(Context):

    def __init__(self, ctx):
        self.gc = ctx.gc

    def rds_frame():
        tbl = self.gc.create_dynamic_frame.from_options(
                    connection_type = "sqlserver", 
                    connection_options = {"url": "jdbc:sqlserver://blablabla.rds.amazonaws.com:1433/dwh", 
                                            "user": "", 
                                            "password": "",
                                            "dbtable": "dbo.tble_name"},
                    transformation_ctx = "glue_df")
        return tbl.printSchema()

def main():

    job = Context()

    args = job.get_parameters()
    metadata = args["dynamodb_metadata_table"]

    RDSClient.rds_frame(job)


if __name__ == "__main__":
    main()

原代码核心问题 #

1. SparkContext初始化错误：self.sc = SparkContext()会直接创建新实例，后续调用getOrCreate()完全冗余，正确用法是通过类方法SparkContext.getOrCreate()获取/创建实例。
2. 继承关系冗余：RDSClient继承了Context，但实际通过__init__接收外部ctx实例，继承未发挥作用反而增加耦合。
3. 实例方法定义错误：rds_frame()未声明self参数，却在方法内使用self，调用时用类直接调用并传参，不符合Python类方法规范。
4. 静态方法调用不规范：用实例job.get_parameters()调用静态方法，虽可行但不如直接用Context.get_parameters()清晰。

优化方案1：组合模式（推荐） #

采用组合而非继承，让RDSClient持有Context实例，保持职责单一：

import sys
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext

class Context:
    def __init__(self):
        # 正确使用getOrCreate获取共享SparkContext
        self.sc = SparkContext.getOrCreate()
        self.gc = GlueContext(self.sc)
        self.spark = self.gc.spark_session
    
    @staticmethod
    def get_parameters():
        return getResolvedOptions(sys.argv, ["dynamodb_metadata_table"])

class RDSClient:
    # 通过构造函数注入Context实例，依赖关系清晰
    def __init__(self, context: Context):
        self.context = context

    # 修复实例方法参数，添加self
    def rds_frame(self):
        tbl = self.context.gc.create_dynamic_frame.from_options(
            connection_type="sqlserver",
            connection_options={
                "url": "jdbc:sqlserver://blablabla.rds.amazonaws.com:1433/dwh",
                "user": "",
                "password": "",
                "dbtable": "dbo.tble_name"
            },
            transformation_ctx="glue_df"
        )
        tbl.printSchema()
        return tbl

def main():
    job_context = Context()
    args = Context.get_parameters()
    metadata = args["dynamodb_metadata_table"]
    
    # 初始化RDSClient并传入Context实例
    rds_client = RDSClient(job_context)
    rds_client.rds_frame()

if __name__ == "__main__":
    main()

优化点说明 #

• 职责单一：Context仅负责初始化和管理Spark/Glue上下文，RDSClient仅处理RDS相关操作，避免耦合。
• 正确共享上下文：SparkContext.getOrCreate()确保全局复用同一个Spark实例，符合需求。
• 代码规范：修复实例方法参数，依赖传递清晰，便于测试和扩展。

优化方案2：单例模式（全局共享场景） #

如果作业全程只需要一个Spark/Glue上下文实例，可使用单例模式简化调用：

import sys
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext

class Context:
    _instance = None

    def __new__(cls):
        # 单例逻辑：仅创建一次实例
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.sc = SparkContext.getOrCreate()
            cls._instance.gc = GlueContext(cls._instance.sc)
            cls._instance.spark = cls._instance.gc.spark_session
        return cls._instance
    
    @staticmethod
    def get_parameters():
        return getResolvedOptions(sys.argv, ["dynamodb_metadata_table"])

class RDSClient:
    def __init__(self):
        # 直接获取单例Context，无需手动传递
        self.context = Context()

    def rds_frame(self):
        tbl = self.context.gc.create_dynamic_frame.from_options(
            connection_type="sqlserver",
            connection_options={
                "url": "jdbc:sqlserver://blablabla.rds.amazonaws.com:1433/dwh",
                "user": "",
                "password": "",
                "dbtable": "dbo.tble_name"
            },
            transformation_ctx="glue_df"
        )
        tbl.printSchema()
        return tbl

def main():
    args = Context.get_parameters()
    metadata = args["dynamodb_metadata_table"]
    
    rds_client = RDSClient()
    rds_client.rds_frame()

if __name__ == "__main__":
    main()

适用场景 #

• 作业全程仅需一个Spark/Glue上下文实例。
• 避免在多个模块间传递Context实例，简化代码调用。

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