DataFrame

Package: flyte.io

A Flyte meta DataFrame object, that wraps all other dataframe types (usually available as plugins, pandas.DataFrame and pyarrow.Table are supported natively, just install these libraries).

Known eco-system plugins that supply other dataframe encoding plugins are,

1. flyteplugins-polars - pl.DataFrame
2. flyteplugins-spark - pyspark.DataFrame

You can add other implementations by extending following flyte.io.extend.

The Flyte DataFrame object serves 2 main purposes:

1. Interoperability between various dataframe objects. A task can generate a pandas.DataFrame and another task can accept a flyte.io.DataFrame, which can be converted to any dataframe.
2. Allows for non materialized access to DataFrame objects. So, for example you can accept any dataframe as a flyte.io.DataFrame and this is just a reference and will not materialize till you force .all() or .iter() etc

class DataFrame(
    uri: typing.Optional[str],
    format: typing.Optional[str],
    hash: typing.Optional[str],
)

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Properties

Methods

all()

def all()

all_sync()

def all_sync()

column_names()

def column_names()

columns()

def columns()

deserialize_dataframe()

def deserialize_dataframe(
    info,
) -> DataFrame

from_df()

def from_df(
    val: typing.Optional[typing.Any],
    uri: typing.Optional[str],
) -> DataFrame

Deprecated: Please use wrap_df, as that is the right name.

Creates a new Flyte DataFrame from any registered DataFrame type (For example, pandas.DataFrame). Other dataframe types are usually supported through plugins like flyteplugins-polars, flyteplugins-spark etc.

from_existing_remote()

def from_existing_remote(
    remote_path: str,
    format: typing.Optional[str],
    kwargs,
) -> 'DataFrame'

Create a DataFrame reference from an existing remote dataframe.

from_local()

def from_local(
    df: typing.Any,
    columns: typing.OrderedDict[str, type[typing.Any]] | None,
    remote_destination: str | None,
    hash_method: HashMethod | str | None,
) -> DataFrame

This method is useful to upload the dataframe eagerly and get the actual DataFrame.

This is useful to upload small local datasets onto Flyte and also upload dataframes from notebooks. This uses signed urls and is thus not the most efficient way of uploading.

In tasks (at runtime) it uses the task context and the underlying fast storage sub-system to upload the data.

At runtime it is recommended to use DataFrame.wrap_df as it is simpler.

Example (With hash_method for cache key computation):

import pandas as pd
from flyte.io import DataFrame, HashFunction

def hash_pandas_dataframe(df: pd.DataFrame) -> str:
    return str(pd.util.hash_pandas_object(df).sum())

@env.task
async def foo() -> DataFrame:
    df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
    hash_method = HashFunction.from_fn(hash_pandas_dataframe)
    return await DataFrame.from_local(df, hash_method=hash_method)

from_local_sync()

def from_local_sync(
    df: typing.Any,
    columns: typing.OrderedDict[str, type[typing.Any]] | None,
    remote_destination: str | None,
    hash_method: HashMethod | str | None,
) -> DataFrame

This method is useful to upload the dataframe eagerly and get the actual DataFrame.

This is useful to upload small local datasets onto Flyte and also upload dataframes from notebooks. This uses signed urls and is thus not the most efficient way of uploading.

In tasks (at runtime) it uses the task context and the underlying fast storage sub-system to upload the data.

At runtime it is recommended to use DataFrame.wrap_df as it is simpler.

Example (With hash_method for cache key computation):

import pandas as pd
from flyte.io import DataFrame, HashFunction

def hash_pandas_dataframe(df: pd.DataFrame) -> str:
    return str(pd.util.hash_pandas_object(df).sum())

@env.task
def foo() -> DataFrame:
    df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
    hash_method = HashFunction.from_fn(hash_pandas_dataframe)
    return DataFrame.from_local_sync(df, hash_method=hash_method)

iter()

def iter()

model_post_init()

def model_post_init(
    context: Any,
)

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

open()

def open(
    dataframe_type: Type[DF],
)

Load the handler if needed. For the use case like: @task def t1(df: DataFrame): import pandas as pd df.open(pd.DataFrame).all()

pandas is imported inside the task, so panda handler won’t be loaded during deserialization in type engine.

schema_match()

def schema_match(
    incoming: dict,
) -> bool

serialize_dataframe()

def serialize_dataframe()

set_literal()

def set_literal(
    expected: types_pb2.LiteralType,
)

A public wrapper method to set the DataFrame Literal.

This method provides external access to the internal _set_literal method.

wrap_df()

def wrap_df(
    val: typing.Optional[typing.Any],
    uri: typing.Optional[str],
) -> DataFrame

Wrapper to create a DataFrame from a dataframe. Other dataframe types are usually supported through plugins like flyteplugins-polars, flyteplugins-spark etc.