Spark 中的声明式数据管道 | 类和装饰器

# Spark 中使用类和装饰器构建数据管道的声明式方法

在数据工程项目中，由于缺乏统一架构，业务逻辑往往会变得碎片化。所提出方法利用 Python 元编程创建声明式管道，其中数据转换使用类和装饰器定义。这种方法简化了维护，提高了代码可重用性，并使业务流程显式化。

声明式管道的核心原则

关键思想是将业务流程（Flow）视为主要实体，而将表视为辅助对象。这可以防止系统扩展时逻辑稀释。Flow 由一系列步骤组成，每个步骤：

• 接收输入表
• 执行转换
• 返回结果

重要的是，表作为类实现，而不是主要对象。例如，MyTable 类背后可能是一个 Spark DataFrame。这种抽象级别允许更改底层实现而无需重写业务逻辑。步骤必须保持原子性，具有明确的输入和输出，以确保可预测的执行。

使用类和装饰器的实现

基础 Flow 类容纳框架逻辑。具体管道通过继承创建：

class MyFlow(Flow):
    @classmethod
    @Flow.step(order=1)
    @Flow.input([MyTable])
    @Flow.output(MyTable2)
    def step_one(cls, context: Context) -> DataFrame:
        print(f" Step 1: sozdayom/update MyTable,{context.id}")

装饰器不执行代码；它们累积元数据：

• @Flow.step(order=1) — 定义执行顺序
• @Flow.input() — 指定输入表
• @Flow.output() — 设置输出

当创建子类时，__init_subclass__ 方法启动，自动收集步骤，按顺序排序，并构建可执行序列。这是通过检查类属性并从装饰方法中提取元数据完成的。

状态管理：Context 和表

步骤之间使用 Context 对象传递数据，该对象实现为 Pydantic 模型：

class Context(BaseModel):
    config: Dict[Type[Config], Config] = {}
    data: Dict[Union[str, Type[Table]], DataFrame] = {}
    diff: Dict[Any, Any] = {}

这种方法的好处：

• 最小化方法参数
• 提供集中式状态管理
• 为扩展提供灵活性

表使用 SQLAlchemy 的声明式风格定义。转换器将 ORM 模型转换为 Spark schema：

class Table(Base):
    __abstract__ = True

    @classmethod
    def get_schema(cls) -> T.StructType:
        fields = []
        for column in cls.__table__.columns:
            spark_type = _convert_type(column.type)
            fields.append(
                T.StructField(column.name, spark_type, column.nullable)
            )
        return T.StructType(fields)

这使得系统各层具有统一的数据模型。如果需要，可以直接在 get_schema() 方法中覆盖 schema。

运行管道和 SQL 处理

管道执行通过 run() 方法按指定顺序顺序调用步骤。推荐 SQL 作为转换的主要工具：

class MyFlow(Flow):
    @classmethod
    @Flow.step(order=3)
    @Flow.input([MyTable])
    @Flow.output(AnotherTable)
    @Flow.sql("step_three.sql")
    def step_three(cls, context: Context):
        df = cls.execute_sql(context, "step_three.sql", vars={"id": 1})

@Flow.sql 装饰器准备参数化查询。在执行前，SQL 使用 create_temp_views() 方法自动为输入表创建临时视图。这允许在查询中无缝使用表，而无需手动处理上下文。

关键点

• 声明式风格使业务逻辑显式化并简化调试
• Context 作为状态的单一访问点减少错误
• 优先使用 SQL 而非 DataFrame API 加速开发并提高可读性
• 通过 __init_subclass__ 的元编程自动化管道组装

— Editorial Team