직렬화(Serialization) 해석 순서와 커스텀 직렬화기 작성
Apache Airflow Official Documentation (in-repo snapshot) — Apache Software Foundation authoring-and-scheduling/serializers.rst - Serialization, Serialization resolution order, Airflow Object, Registered (전체, 144줄)
이 모듈을 다 읽으면
- Airflow가 값을 직렬화할 때 시도하는 4단계 해석 순서를 설명할 수 있다
- 자신이 통제하는 클래스와 통제하지 않는 클래스에 각각 어떤 직렬화 방식을 써야 하는지 판단할 수 있다
- 등록형(Registered) 직렬화기 작성 시 순환 참조를 피하는 방법을 설명할 수 있다
태스크 간 값 전달(XCom 등)과 웹서버/스케줄러가 Dag 파일을 직접 읽지 않아도 되게 하기 위해 Airflow는 값을 직렬화하는데, 원시값→등록된 직렬화기→객체의 serialize()/deserialize() 메서드→@dataclass/@attr.define 순으로 해석하며, 통제 가능한 클래스는 serialize/deserialize를, 통제 불가능한 외부 클래스는 등록형 직렬화기를 쓰는 것이 권장된다.
왜 직렬화가 필요한가
태스크 사이에 인자 같은 데이터를 주고받으려면 Airflow는 그 값을 직렬화했다가, 다운스트림 태스크에서 필요할 때 다시 역직렬화해야 한다. 직렬화는 또한 웹서버와 스케줄러(Dag 프로세서와 달리)가 보안과 효율을 위해 Dag 파일을 직접 읽을 필요가 없게 만드는 목적도 있다.
직렬화는 의외로 까다로운 작업이다. 파이썬은 기본적으로 `str`, `int` 같은 원시값과 그런 원시값들의 iterable만 자연스럽게 다루며, 그보다 복잡한 객체가 등장하면 커스텀 직렬화가 필요해진다.
핵심 포인트
- 태스크 간 값 교환뿐 아니라, 웹서버와 스케줄러가 보안·효율을 위해 Dag 파일을 직접 읽지 않고도 값을 다룰 수 있게 하려고 직렬화가 필요하다
- 파이썬은 기본적으로 str/int 같은 원시값과 그 원시값들의 iterable만 자연스럽게 다루며, 복잡한 객체는 커스텀 직렬화가 필요하다
직렬화 해석 순서 4단계
Airflow는 커스텀 직렬화를 다음 순서로 해석한다.
첫째, `str`이나 `int` 같은 원시값과 원시값들의 iterable은 추가 인코딩 없이 그대로 반환된다. 둘째, 원시값이 아니라면 `airflow.sdk.serde.serializers` 네임스페이스에 등록된 직렬화기와 역직렬화기가 있는지 찾는다. 셋째, 등록된 직렬화기가 없다면 그 객체가 `serialize()` 메서드(그리고 역직렬화를 위한 `deserialize(data, version: int)` 메서드)를 정의하고 있는지 확인한다. 넷째, 마지막으로 객체가 `@dataclass`나 `@attr.define`으로 데코레이트되어 있다면, 그 데코레이터가 제공하는 공개 필드를 이용해 직렬화한다.
이 순서는 곧 어떤 방식을 선택할지 판단하는 기준이기도 하다. Airflow가 통제하는 객체(예: `airflow.model.dag.DAG`)나 개발자가 통제하는 객체(예: `my.company.Foo`)는 먼저 `@attr.define`이나 `@dataclass`로 데코레이트할 수 있는지 검토하고, 그럴 수 없다면 `serialize`/`deserialize` 메서드를 구현하는 것이 좋다. 반면 Airflow가 통제하지 않는 객체(예: `numpy.int16`)는 등록형 직렬화기와 역직렬화기가 필요하다.
핵심 포인트
- 해석 순서는 원시값(그대로 반환) → airflow.sdk.serde.serializers 네임스페이스에 등록된 직렬화기 → 객체 자신의 serialize()/deserialize() 메서드 → @dataclass/@attr.define 공개 필드 순이다
- 이 순서는 곧 어떤 방식을 쓸지 선택 기준이기도 하다 - Airflow가 통제하는 객체나 개발자가 통제하는 객체는 먼저 @attr.define/@dataclass나 serialize/deserialize 메서드를 검토하고, numpy.int16처럼 통제 밖의 객체는 등록형 직렬화기가 필요하다
serialize/deserialize 메서드로 직접 구현하기
통제 가능한 객체를 직렬화하는 가장 단순한 형태는 클래스에 `serialize()`와 정적 메서드 `deserialize(data, version)`를 직접 구현하는 것이다. `serialize()`는 원시값이나 dict를 반환하면 된다 - dict의 값 자체를 미리 직렬화할 필요는 없고(Airflow가 알아서 처리한다) 다만 dict의 키는 반드시 원시 형태여야 한다.
클래스에 `__version__`(ClassVar[int])을 두고, `deserialize(data, version)`에서 이 버전 정보를 이용해 호환성을 스스로 검사하도록 만드는 패턴이 권장된다.
핵심 포인트
- serialize()는 원시값이나 dict를 반환하면 되고, dict의 값 자체를 미리 직렬화할 필요는 없지만(자동 처리됨) 키는 반드시 원시 형태여야 한다
- 클래스에 __version__(ClassVar[int])을 두고 deserialize(data, version)에서 버전 호환성을 스스로 검사하는 패턴이 권장된다
등록형(Registered) 직렬화기와 순환 참조 방지
Airflow가 통제하지 않는 외부 객체(예: `decimal.Decimal`, `numpy.int16`)는 버저닝을 갖춘 등록형 직렬화기와 역직렬화기가 필요하다. 이런 모듈은 `serializers`/`deserializers`라는 클래스패스(또는 타입) 리스트와 `__version__`을 모듈 레벨에 선언하고, `serialize(o)`는 `(직렬화된 값, 클래스명, 버전, 직렬화 성공 여부)` 튜플을, `deserialize(cls, version, data)`는 `cls`/`version`/`data`를 받아 원래 객체를 복원해 반환한다. `deserialize`에서는 항상 저장된 `version`이 현재 코드의 `__version__`보다 크지 않은지 검사해, 미래 버전의 포맷을 구버전 코드가 잘못 해석하는 상황을 막아야 한다.
순환 참조(circular import)를 피하려면, 직렬화 대상 클래스를 직접 import해 `serializers = [Foo]`처럼 쓰는 대신 `serializers = ["my.company.Foo"]`처럼 클래스를 문자열 경로로 등록하고, 실제 클래스는 직렬화기 함수 내부에서 지연 import하는 것이 안전하다.
속도 측면에서, 직렬화/역직렬화는 속도에 민감한 작업이므로 가능한 한 `dict` 같은 내장 자료형을 활용하고 클래스나 다른 복잡한 구조는 피하는 것이 좋다.
핵심 포인트
- Airflow가 통제하지 않는 외부 객체(예: decimal.Decimal, numpy.int16)는 버저닝을 갖춘 등록형 직렬화기/역직렬화기가 필요하다 - serialize는 (값, 클래스명, 버전, 직렬화여부) 튜플을, deserialize는 cls/version/data를 받아 원래 객체를 복원한다
- 순환 참조를 피하려면 serializers = ["my.company.Foo"]처럼 클래스를 문자열 경로로 등록하고, 실제 클래스는 직렬화기 함수 내부에서 지연 import하는 것이 안전하다
- deserialize에서는 항상 저장된 version이 현재 코드의 __version__보다 크지 않은지 검사해, 향후 버전에서 저장된 데이터를 구버전 코드가 잘못 해석하지 않도록 해야 한다