Task Output Attributes

[SandraGH5]

How can I access the attributes of task outputs to be used as inputs to other tasks? For example, below I use a dataclass:

@dataclass_json
@dataclass
class Person:
    name: str
    nationality: str

@task
def create_person(name: str, nationality: str) -> Person:
    return Person(name=name, nationality=nationality)

@task
def say_hello(name: str) -> str:
    greeting = f"hello {name}"
    print(greeting)
    return greeting

@workflow
def my_wf(name: str="aria", nationality:str="Indonesian") -> str:
    person = create_person(name=name, nationality=nationality)
    return say_hello(name=person.name). --> throws AttributeError: 'Promise' object has no attribute 'name'

• Pradithya Aria Pura

[SandraGH5]

Currently, dataclass converts the object to a flyteidl.struct / json type. And today flytepropeller cannot open up the struct to pass parts of it to downstream tasks. (This is a feature we have been thinking about, but are afraid that the engine will need to load arbitrary jsons).
Thus you can only pass dataclasses as one complete object and cannot access individual variables from it. You can change your code to:

Person = typing.NamedTuple(
    "Person", name=str, nationality=str)

@task
def create_person(name: str, nationality: str) -> Person:
    return Person(name=name, nationality=nationality)

@task
def say_hello(name: str) -> str:
    greeting = f"hello {name}"
    print(greeting)
    return greeting

@workflow
def my_wf(name: str="aria", nationality:str="Indonesian") -> str:
    person = create_person(name=name, nationality=nationality)
    return say_hello(name=person.name)

This should work, and the reason is, this is treated as multiple outputs (tuple), just named.

Here is where the tuple opening happens:
This is the function that is called, when you invoke the task and the behavior depends on the state.
A task state can be one of 3:

1. Compiling (i.e.) only called for workflow creation and never ends up executing the function
2. Dynamic execute (which is 2 step - compile + execute)
3. Actual execute, where the inputs are unwrapped and passed in, and the outputs are wrapped

https://github.com/flyteorg/flytekit/blob/master/flytekit/core/base_task.py#L273

At runtime (in production, when the cli is invoked), the big call part is bypassed and directly dispatch_execute is invoked. These lines show how to unwrap and handle various outputs:
https://github.com/flyteorg/flytekit/blob/master/flytekit/core/base_task.py#L464-L517.

Replying to:
"So for the case that I have to use dataclass or other user-defined class, I’ll have to either pass it as a whole or create a dedicated task to unwrap it right?"
Yes, and just in case you are thinking how to pass legacy non-dataclass python objects, we just released v0.23.0, which brings with it PickleType support for arbitrary python objects.

flyteorg/flytekit#667

USE WITH CAUTION! This only works across non-shared tasks.

In Flyte, you can actually reference tasks from other projects and create your own workflow that uses tasks that are declared by others.
These tasks can be in their own containers, own python runtimes, their own flytekit versions and may even be in different languages. The only thing that you need is the interface to connect them with your tasks. This is like calling an RPC service.
This can be done using:

@reference_task(project, domain, name, optional-version)
def foo( input interface ) -> outputs:
     ...

We and Spotify and even Freenome use it to create platform tasks that others can simply re-use.
Here's an example:
https://docs.flyte.org/projects/cookbook/en/latest/auto/core/flyte_basics/reference_task.html

@kumare3
@samhita-alla

[main2adwitiya]

This is zz best comment ever, saved us a huge chunk of time.

[yubofredwang]

"USE WITH CAUTION! This only works across non-shared tasks."

Just wondering how you handle object passing if you want to use it with Task Reference?

[samhita-alla]

Sorry, what do you mean by task reference?

[yubofredwang]

I am referring to this one:
https://docs.flyte.org/projects/cookbook/en/latest/auto/core/flyte_basics/reference_task.html#reference-task

[samhita-alla]

This has to work with reference tasks as well. Are you seeing any issues?