Python中如何实现普通类与Apache Beam DoFn/ParDo类的互相调用继承

Python不支持双向继承（循环继承会直接触发解释器类定义错误），且Apache Beam的DoFn类有专属的生命周期、序列化分发规则，优先用组合逻辑实现两类的互相调用，稳定性远高于硬写继承。

方案1：组合模式（生产环境优先推荐） #

这个方案完全规避Beam序列化坑、类继承冲突问题，两类逻辑完全解耦，维护成本最低。

1.1 在DoFn类中调用普通generic类的方法 #

不要在DoFn的__init__构造方法中初始化generic实例（构造方法在Pipeline提交节点执行，生成的实例需要序列化后分发到计算节点，容易触发不可序列化对象报错），要在Beam DoFn自带的setup()生命周期方法中初始化generic实例，setup()会在每个计算节点加载DoFn时执行，没有序列化问题。
代码示例：

# beam.py
import apache_beam as beam
from generic import generic

class rejected_records(beam.DoFn):
    def setup(self):
        # 计算节点本地初始化普通类实例
        self.generic_handler = generic()

    def process(self, element):
        """
        Transformation
        """
        # 按需调用generic类的a/b/c方法
        res_a = self.generic_handler.a(element)
        res_b = self.generic_handler.b(res_a)
        res_c = self.generic_handler.c(res_b)
        yield res_c

1.2 在普通generic类中调用DoFn的转换逻辑 #

DoFn本身是为Beam ParDo设计的执行单元，脱离Beam Pipeline上下文直接实例化调用process方法不会触发完整生命周期，容易出异常。把DoFn中不依赖Beam运行上下文的核心转换逻辑抽为静态方法，普通类可以直接导入调用，DoFn的process方法也能复用这段逻辑。
代码示例：

# 改造后的beam.py
import apache_beam as beam
from generic import generic

class rejected_records(beam.DoFn):
    @staticmethod
    def core_transform(element):
        # 抽离核心转换逻辑，不依赖Beam上下文
        """Transformation 核心逻辑"""
        return element

    def setup(self):
        self.generic_handler = generic()

    def process(self, element):
        transformed = self.core_transform(element)
        yield self.generic_handler.c(transformed)

# 改造后的generic.py
from beam import rejected_records

class generic():
    def a(self, val):
        # 方法a逻辑
        return val
    def b(self, val):
        # 方法b逻辑
        return val
    def c(self, val):
        # 直接调用DoFn抽离的核心转换逻辑
        return rejected_records.core_transform(val)

方案2：Mixin多继承（适合无状态工具类场景） #

如果generic类是无状态的工具类（没有持有数据库连接、文件句柄这类不可序列化的资源），可以把generic类改造成Mixin类，让DoFn类多继承Mixin，直接复用a/b/c方法，不需要单独初始化实例。
注意：禁止反向让generic类继承DoFn，普通类没有Beam运行上下文，继承DoFn没有实际意义还会徒增依赖。
代码示例：

# generic.py 改造成Mixin类
class GenericMixin:
    def a(self, val):
        return val
    def b(self, val):
        return val
    def c(self, val):
        return val

# beam.py
import apache_beam as beam
from generic import GenericMixin

class rejected_records(beam.DoFn, GenericMixin):
    def process(self, element):
        # 直接调用继承得到的a/b/c方法
        step1 = self.a(element)
        step2 = self.b(step1)
        yield self.c(step2)

注意事项 #

• 不要尝试实现双向继承，Python解释器会直接抛出循环继承错误，完全无法运行。
• DoFn的生命周期（setup/process/teardown）由Beam Runner统一管控，不要脱离Pipeline上下文手动实例化DoFn执行业务逻辑，会出现资源未释放、上下文缺失等问题。
• 所有需要分发到计算节点的DoFn持有的对象，必须支持pickle序列化，否则Pipeline提交时会直接报错。

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