Fallback Behavior: Pickling

When a task or workflow in flyte-sdk consumes or outputs a Python object whose type is not explicitly supported by a registered type transformer, Flyte must choose between failing execution at compilation/runtime or falling back to a universal serialization strategy. Flyte resolves unmapped types by converting them to FlytePickle, serializing the object graph using cloudpickle and uploading the resulting byte stream as an offloaded blob.

Understanding this fallback behavior is critical for designing robust pipelines, as pickling provides flexibility at the expense of polyglot interoperability, Flyte Console introspection, and cross-runtime compatibility.

The FlytePickle Serialization Pipeline

The core pickling machinery is implemented in flyte.types._pickle through two collaborating classes: FlytePickle and FlytePickleTransformer.

FlytePickle is a generic container type that provides asynchronous serialization (to_pickle) and deserialization (from_pickle) utilities. It interfaces with cloudpickle for object serialization and flyte.storage for blob transport.

class FlytePickle(typing.Generic[T]):
    """
    This type is only used by flytekit internally. User should not use this type.
    Any type that flyte can't recognize will become FlytePickle
    """

    @classmethod
    def python_type(cls) -> typing.Type:
        return type(None)

    @classmethod
    def __class_getitem__(cls, python_type: typing.Type) -> typing.Type:
        if python_type is None:
            return cls

        class _SpecificFormatClass(FlytePickle):
            __origin__ = FlytePickle

            @classmethod
            def python_type(cls) -> typing.Type:
                return python_type

        return _SpecificFormatClass

    @classmethod
    async def to_pickle(cls, python_val: typing.Any) -> str:
        h = hashlib.md5()
        str_bytes = cloudpickle.dumps(python_val)
        h.update(str_bytes)

        uri = storage.get_random_local_path(file_path_or_file_name=h.hexdigest())
        os.makedirs(os.path.dirname(uri), exist_ok=True)
        async with aiofiles.open(uri, "w+b") as outfile:
            await outfile.write(str_bytes)

        return await storage.put(str(uri))

    @classmethod
    async def from_pickle(cls, uri: str) -> typing.Any:
        if storage.is_remote(uri):
            local_path = storage.get_random_local_path()
            await storage.get(uri, str(local_path), False)
            uri = str(local_path)
        async with aiofiles.open(uri, "rb") as infile:
            data = cloudpickle.loads(await infile.read())
        return data

When an object is serialized:

1. cloudpickle.dumps(python_val) serializes the Python object into bytes.
2. An MD5 hash of the payload bytes is generated via hashlib.md5() and used as the unique local filename returned by storage.get_random_local_path.
3. The serialized bytes are written asynchronously to disk using aiofiles.
4. storage.put(str(uri)) transfers the file to the configured remote blob store (such as S3, GCS, or local storage in local mode), returning a remote storage URI.

During deserialization:

1. FlytePickle.from_pickle(uri) checks storage.is_remote(uri). If remote, it downloads the payload to a local temporary path via storage.get.
2. The file is opened asynchronously with aiofiles.open(..., "rb") and loaded back into memory via cloudpickle.loads.

TypeEngine Fallback Resolution

FlytePickleTransformer bridges FlytePickle to Flyte's IDL representation. It translates Python objects into literals_pb2.Literal instances containing a single-dimensional blob of format "PythonPickle".

class FlytePickleTransformer(TypeTransformer[FlytePickle]):
    PYTHON_PICKLE_FORMAT = "PythonPickle"

    def __init__(self):
        super().__init__(name="FlytePickle", t=FlytePickle)

    def assert_type(self, t: Type[T], v: T):
        # Every type can serialize to pickle, so we don't need to check the type here.
        ...

    async def to_python_value(self, lv: literals_pb2.Literal, expected_python_type: Type[T]) -> T:
        uri = lv.scalar.blob.uri
        return await FlytePickle.from_pickle(uri)

    async def to_literal(
        self,
        python_val: T,
        python_type: Type[T],
        expected: types_pb2.LiteralType,
    ) -> literals_pb2.Literal:
        if python_val is None:
            raise AssertionError("Cannot pickle None Value.")
        meta = literals_pb2.BlobMetadata(
            type=types_pb2.BlobType(
                format=self.PYTHON_PICKLE_FORMAT, dimensionality=types_pb2.BlobType.BlobDimensionality.SINGLE
            )
        )
        remote_path = await FlytePickle.to_pickle(python_val)
        return literals_pb2.Literal(scalar=literals_pb2.Scalar(blob=literals_pb2.Blob(metadata=meta, uri=remote_path)))

    def get_literal_type(self, t: Type[T]) -> types_pb2.LiteralType:
        lt = types_pb2.LiteralType(
            blob=types_pb2.BlobType(
                format=self.PYTHON_PICKLE_FORMAT, dimensionality=types_pb2.BlobType.BlobDimensionality.SINGLE
            )
        )
        lt.metadata = {"python_class_name": str(t)}
        return lt

When TypeEngine.get_transformer(python_type) encounters a type in flyte/types/_type_engine.py, it evaluates type annotations, primitives, registered transformers, subclasses via the Method Resolution Order (MRO), union types, and dataclasses. If no match is found, it calls display_pickle_warning and returns a new FlytePickleTransformer:

if dataclasses.is_dataclass(python_type):
    return cls._DATACLASS_TRANSFORMER

display_pickle_warning(str(python_type))
from flyte.types._pickle import FlytePickleTransformer

return FlytePickleTransformer()

The warning notifies developers about runtime portability implications:

@lru_cache
def display_pickle_warning(python_type: str):
    logger.warning(
        f"Unsupported Type {python_type} found, Flyte will default to use PickleFile as the transport. "
        f"Pickle can only be used to send objects between the exact same version of Python, "
        f"and we strongly recommend to use python type that flyte support."
    )

Additional Pickling Integrations

Pickling is also leveraged in two specialized subsystems within flyte-sdk:

Dictionary Serialization Fallback

In DictTransformer (flyte/types/_type_engine.py), dictionary inputs and outputs default to binary MessagePack encoding (dict_to_binary_literal). If a dictionary contains non-standard Python objects that cause MessagePack encoding to raise a TypeError, Flyte checks whether pickling is allowed for that type hint (e.g. annotated dictionary types evaluated by DictTransformer.is_pickle). If allow_pickle is True, it falls back to FlytePickle.to_pickle(v), packaging the remote path into a Struct literal with {"format": "pickle"}.

CLI Dynamic Import Resolution

When tasks accepting pickled parameters are invoked via Flyte's CLI (flyte/cli/_params.py), Click uses PickleParamType when lt.blob.format == FlytePickleTransformer.PYTHON_PICKLE_FORMAT. Because raw pickled bytes cannot be passed directly on the command line, PickleParamType.convert expects a string formatted as <Module>:<Object> (for example, my_module:my_object). It dynamically imports the module via importlib.import_module and retrieves the variable with getattr.

Trade-offs and Design Implications

Using FlytePickle as a universal fallback offers significant benefits alongside structural constraints.

Advantages

• Zero Configuration for Custom Classes: Arbitrary Python objects—such as third-party library models, custom class instances, and complex nested closures—can be passed between tasks without writing custom TypeTransformer implementations.
• Support for Cyclic and Complex Graphs: cloudpickle can serialize nested object graphs, functions, and lambda expressions that standard JSON, Protobuf, or MessagePack schemas cannot represent.
• Asynchronous Storage Management: Large serialized objects are automatically written to disk and uploaded to remote object storage without blocking event loops.

Disadvantages

• Runtime and Python Version Lock-in: cloudpickle bytecode is sensitive to differences in Python minor versions and library dependency environments. If an upstream task serializing a pickle runs in a Python 3.11 container and the downstream task runs in Python 3.10, deserialization may fail with UnpicklingError or AttributeError.
• UI and Console Opacity: Pickled objects are stored as opaque PythonPickle blobs. Unlike Flyte dataclasses, typed dictionaries, or DataFrames, the Flyte Console cannot inspect, render, filter, or validate the inner contents of a pickled scalar.
• Polyglot Boundary Incompatibility: Pickled blobs are native to Python. Tasks implemented in non-Python SDKs (such as Java, C++, or TypeScript) or external microservices consuming Flyte inputs/outputs cannot parse PythonPickle literals.
• Storage and I/O Overhead: Every pickled input or output requires serializing memory state, calculating an MD5 hash, writing to local storage, and issuing remote upload/download operations. For structured tabular data, FlyteFile, FlyteDirectory, or native DataFrame types provide higher throughput and better memory efficiency.
• Rejection of None Values: FlytePickleTransformer.to_literal explicitly raises AssertionError("Cannot pickle None Value."). None values must instead be handled through native SimpleType.NONE bindings.