如何让父类定义的实例属性在流水线子类步骤中持久生效

问题根因 #

当前代码中root_path是InitialiseStep的实例属性，不同实例之间属性相互独立。你创建的InitialiseStep实例和后续的CampaignData实例是两个完全独立的对象，后者的root_path从未被赋值，默认继承父类的初始值None，所以会输出空值。

可行解决方案 #

方案1：类属性全局共享（修改量最小） #

把root_path设置为InitialiseStep的类属性，所有继承该类的子类实例都可以共享同一个值，修改后代码如下：

class InitialiseStep():
    # 定义为类属性
    root_path = None
    def __init__(self, root_path=None):
        # 赋值给类属性而不是实例属性
        if root_path is not None:
            InitialiseStep.root_path = root_path
    def execute(self):
        print(self.root_path)

class SparkStep(InitialiseStep):
    def __init__(self, spark=None):
        super().__init__()
        self.spark = spark
    def execute(self):
        print(self.root_path)

class CampaignData(SparkStep):
    def __init__(self, spark):
        super().__init__()
        self.spark = spark

class Executor:
    def __init__(self, steps):
        self.steps = steps
    def run_step(self, step):
        step.execute()
    def run_pipeline(self):
        for step in self.steps:
            self.run_step(step)

# 补充原代码缺失的spark变量定义
spark = None
executor = Executor([InitialiseStep(root_path="/test/path/"), CampaignData(spark)])
executor.run_pipeline()

运行输出：

/test/path/
/test/path/

方案2：流水线上下文传递（更规范，适合复杂流水线场景） #

如果后续流水线需要共享的配置不止root_path一个，建议引入全局上下文对象，所有步骤初始化时统一传入上下文，实现配置共享：

# 全局上下文类，存储全流水线共享配置
class PipelineContext:
    def __init__(self, root_path=None, spark=None):
        self.root_path = root_path
        self.spark = spark

class InitialiseStep():
    def __init__(self, context):
        self.context = context
    def execute(self):
        print(self.context.root_path)

class SparkStep(InitialiseStep):
    def __init__(self, context):
        super().__init__(context)
    def execute(self):
        print(self.context.root_path)

class CampaignData(SparkStep):
    def __init__(self, context):
        super().__init__(context)

class Executor:
    def __init__(self, steps):
        self.steps = steps
    def run_step(self, step):
        step.execute()
    def run_pipeline(self):
        for step in self.steps:
            self.run_step(step)

# 初始化全局上下文，统一管理配置
context = PipelineContext(root_path="/test/path/", spark=None)
executor = Executor([InitialiseStep(context), CampaignData(context)])
executor.run_pipeline()

该方案的优势是配置集中管理，不会出现类属性被意外修改影响全流程的问题，扩展性更强。

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