Spark에서의 선언형 데이터 파이프라인 | Classes and Decorators

# 데이터 파이프라인 구축을 위한 선언적 접근 방식: 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 스키마로 변환합니다:

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() 메서드에서 스키마를 직접 오버라이드할 수 있습니다.

파이프라인 실행과 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는 오류를 줄임
• DataFrame API보다 SQL을 우선시하면 개발 속도가 빨라지고 가독성이 향상
• __init_subclass__를 통한 메타프로그래밍이 파이프라인 조립을 자동화

— Editorial Team