Serialization

To support data exchange, like arguments, between tasks, Airflow needs to serialize the data to be exchanged and deserialize it again when required in a downstream task. Serialization also happens so that the webserver and the scheduler (as opposed to the DAG processor) do no need to read the DAG file. This is done for security purposes and efficiency.

Serialization is a surprisingly hard job. Python out of the box only has support for serialization of primitives, like str and int and it loops over iterables. When things become more complex, custom serialization is required.

Airflow out of the box supports three ways of custom serialization. Primitives are are returned as is, without additional encoding, e.g. a str remains a str. When it is not a primitive (or iterable thereof) Airflow looks for a registered serializer and deserializer in the namespace of airflow.serialization.serializers. If not found it will look in the class for a serialize() method or in case of deserialization a deserialize(data, version: int) method. Finally, if the class is either decorated with @dataclass or @attr.define it will use the public methods for those decorators.

If you are looking to extend Airflow with a new serializer, it is good to know when to choose what way of serialization. Objects that are under the control of Airflow, i.e. residing under the namespace of airflow.* like airflow.model.dag.DAG or under control of the developer e.g. my.company.Foo should first be examined to see whether they can be decorated with @attr.define or @dataclass. If that is not possible then the serialize and deserialize methods should be implemented. The serialize method should return a primitive or a dict. It does not need to serialize the values in the dict, that will be taken care of, but the keys should be of a primitive form.

Objects that are not under control of Airflow, e.g. numpy.int16 will need a registered serializer and deserializer. Versioning is required. Primitives can be returned as can dicts. Again dict values do not need to be serialized, but its keys need to be of primitive form. In case you are implementing a registered serializer, take special care not to have circular imports. Typically, this can be avoided by using str for populating the list of serializers. Like so: serializers = ["my.company.Foo"] instead of serializers = [Foo].

Note

Serialization and deserialization is dependent on speed. Use built-in functions like dict as much as you can and stay away from using classes and other complex structures.

Airflow Object

  1. from typing import Any, ClassVar
  2. class Foo:
  3. __version__: ClassVar[int] = 1
  4. def __init__(self, a, v) -> None:
  5. self.a = a
  6. self.b = {"x": v}
  7. def serialize(self) -> dict[str, Any]:
  8. return {
  9. "a": self.a,
  10. "b": self.b,
  11. }
  12. @staticmethod
  13. def deserialize(data: dict[str, Any], version: int):
  14. f = Foo(a=data["a"])
  15. f.b = data["b"]
  16. return f

Registered

  1. from __future__ import annotations
  2. from decimal import Decimal
  3. from typing import TYPE_CHECKING
  4. from airflow.utils.module_loading import qualname
  5. if TYPE_CHECKING:
  6. from airflow.serialization.serde import U
  7. serializers = [
  8. Decimal
  9. ] # this can be a type or a fully qualified str. Str can be used to prevent circular imports
  10. deserializers = serializers # in some cases you might not have a deserializer (e.g. k8s pod)
  11. __version__ = 1 # required
  12. # the serializer expects output, classname, version, is_serialized?
  13. def serialize(o: object) -> tuple[U, str, int, bool]:
  14. if isinstance(o, Decimal):
  15. name = qualname(o)
  16. _, _, exponent = o.as_tuple()
  17. if exponent >= 0: # No digits after the decimal point.
  18. return int(o), name, __version__, True
  19. # Technically lossy due to floating point errors, but the best we
  20. # can do without implementing a custom encode function.
  21. return float(o), name, __version__, True
  22. return "", "", 0, False
  23. # the deserializer sanitizes the data for you, so you do not need to deserialize values yourself
  24. def deserialize(classname: str, version: int, data: object) -> Decimal:
  25. # always check version compatibility
  26. if version > __version__:
  27. raise TypeError(f"serialized {version} of {classname} > {__version__}")
  28. if classname != qualname(Decimal):
  29. raise TypeError(f"{classname} != {qualname(Decimal)}")
  30. return Decimal(str(data))