Hi, is it possible to know why it not recommended to drive any production logic on the execution_date ? My use case is the following : I want to schedule a dag every day to load data from an API, the Api require a date, I want to to use the execution date, to me it a safe usage please let me know if I am wrong

Hi, everyone. I'm having some trouble using the sandbox to pass DataFrames between tasks. No IAM roles configured, no Service Accounts, just a task that returns a DataFrame. This is the error I'm having:

File "/home/flyte/.venv/lib/python3.9/site-packages/flytekit/core/data_persistence.py", line 447, in put_data
    raise FlyteAssertion(
flytekit.exceptions.user.FlyteAssertion: Failed to put data from /tmp/flyte-9ibktn0s/sandbox/local_flytekit/engine_dir to <s3://my-s3-bucket/metadata/propeller/search-filter-improvement-by-semantic-search-development-alqv5l4nv4kms759s7k2/n0/data/0> (recursive=True).

Original exception: Called process exited with error code: 1.  Stderr dump:

b'upload failed: ../../tmp/flyte-9ibktn0s/sandbox/local_flytekit/engine_dir/error.pb to <s3://my-s3-bucket/metadata/propeller/search-filter-improvement-by-semantic-search-development-alqv5l4nv4kms759s7k2/n0/data/0/error.pb> An error occurred (AccessDenied) when calling the PutObject operation: Access Denied\n'

Why is this happening? Sandbox image I'm using: cr.flyte.org/flyteorg/flyte-sandbox:dind-106a8147446a6f0221162f47a9260ea0a764426e Signature of the task:

def check_new_amnt(sql_path: str, config_path: str) -> pd.DataFrame:

Hi, is it possible to manually hard delete executions through

flytectl

? I have 3 workflow executions that have been stuck in

ABORTING

for the past ~2 hours.

Question regarding fast-registration and tar archives: I've noticed you recently added this

tar_strip_file_attributes

functions. What is the purpose of that? When unpacking fast-registered archives it messes up the permissions:

d--------- 3 root root 4096 Jul 14 14:04 unpacked
-rw-r--r-- 1 root root  335 Jul 14 13:59 scriptmode.tar.gz

It doesn't matter if you run as root but we deliberately don't in our containers so everything fast-registered with new flytekit currently breaks.

Can I get a quick review of a new 0.26 release just changing some requirements https://github.com/flyteorg/flytekit/pull/1105

Hi, everyone. I was wondering: is there a way to programatically register a package of tasks/workflows? Using

FlyteRemote

and not

flytectl

, that is. My use case is that I want to develop a Python script that executes the entire deployment process, but I do not want to call subprocesses, since that would require the user to install

flytectl

(and some members of our team use Windows!). I know that

FlyteRemote

can register individual tasks and workflows, but is there a way to register an entire package with it?

For visibility of folks ---- There’s a separate issue we’ve noticed with

NamedTuple

vs

dataclass

support. •

@workflow

and

@dynamic

seem to have some special Promise handling for

NamedTuple

types specifically ◦ Within a

@workflow

I can construct

NamedTuple

instances, including passing promise objects into their attributes ◦ Within a

@workflow

I can access attributes on a

NamedTuple

instance and get a promise object back which resolves to a field in that record ◦ The return types of these seem equivalent to the original

NamedTuple

definition, but not the same as, ie.

isinstance(output, MyNamedTuple) == False

on a task output from a task that returns a

MyNamedTuple

, even though the object in all other respects pretends to be a

MyNamedTuple

instance. • None of this behavior is supported for

dataclass

types ◦ Including the wonky

isinstance

stuff Is this special-cased logic for

NamedTuple

? • We’d really love to use exactly one of

NamedTuple

or

dataclass

for all of our structures in our Flyte workflows • The above behaviors are very desirable (in fact I think I asked for exactly this ~6mo ago! very excited to see it) • But the mis-matched support of features for

NamedTuple

and

dataclass

is making it hard for us to put a stake in the ground and just pick one. • Can we either implement the above behaviors (minus the isinstance bit) for

dataclass

or allow

NamedTuple

s as task inputs? ---------- Let me explain what NamedTuple are meant to be. So when you have multi-valued return from a python function, i.e.,

def foo() -> (int, int, int):
   return 10, 10, 10

In this case the

int, int, int

is actually a tuple in the python world right, as you can receive it as

x = foo()

and

x

would be a tuple. But Python does special handing for Tuples as the following is valid

m, n, o = foo()

Flyte actually supports multiple values as outputs. These are not python but arbitrary lanugages. By default, flytekit names all the return types as

o1, o2, …, on

This is because every returned value in Flyte is actually bound by its name

o1

etc in the downstream consumer Flyte also supports naming the outputs - as

x, y, z

instead of the

o1…

But, in python there is no easy way to name the outputs. So we decided to use NamedTuples. Rationale, NamedTuples behave the same as Tuples, but allow naming. NamedTuples are not handled specially in flytekit, except for the naming. On the otherhand, dataclasses are very python native thing. We wanted users to be able to return arbitrarily complex objects as a single value, so in the above example

def foo() -> (int, dataclass, ...)"
   ...

Dataclasses are sent through flyte as opaque Json Objects. And the engine tries to not load them to keep its security high. thus dataclasses are more like

int

tuples and namedtuples

cannot be returned today as an invidual object

def foo() -> (int, tuple)
  ....

Is not valid

Hi! I ran into some issues trying to schedule some worfklow. The docs say I could activate my launch plan using a command

pyflyte lp -p {{ your project }} -d {{ your domain }} activate-all

The thing is,

lp

subcommand does not exist at all, so the command above fails. Ideally would like to do it programmatically from Python having a handle to a launch plan object. Is this possible?

Hi! I’m trying to add a kubernetes pod label to all pods generated as a result of launching a workflow. Execution code:

remote.execute(
        workflow,
        inputs=inputs,
        options=Options(labels=common_models.Labels({"flyte.user": os.getenv("USER", "na")}))

where

remote

is of type

FlyteRemote

. I can see that the workflow’s pod has the expected label, but none of the subworkflows do. Is there a way to have all pod labels be passed to all child pods of a workflow pod?

has there been any consideration to caching at the lp/workflow level?

Is it possible to mix

@flytekit.dynamic

&

with_overrides

? Curious if we could pass a resource configuration object to a workflow at runtime and have the

@dynamic

step specify overrides at runtime

anyone else use flytectl with M1 chips? getting a segfault when running

Cc @Kevin Su / @Yee - the protocol part - https://github.com/flyteorg/flytekit/pull/1116 - is that right? Can we talk and change it? @Evan Sadler

Upgrading to flytekit 1.0 and i have a task taking a single schema which can't marshal it to a dataframe because this conditional is false

lv and lv.scalar and lv.scalar.schema

. looking at the inputs it seems that old versions of the task don't have the

format

field in the flyte literal, but now they do. also looks like this code returns empty dataframe rather than erroring. any context on these changes that might help point to the behavior we're seeing?

side question from the above, a structured datasets and flyteschemas interoperable at all

more flytekit 1.0 questions: is there a way to set default resource settings? we used to use flyte-cli with the flyte conf file but that's no longer used

this seems potentially really cool, but i don't see an explainer in the docs. what does this do?

Annotated[StructuredDataset, subset_cols]

Is there a verbose mode of sorts that we can enable when running flyte workflows? Maybe not always, but when the workflow fails?

Hi, I'm reading the Flyte documentation on extending a task. There are two ways listed.

@task(task_config=MyContainerExecutionTask(
    plugin_specific_config_a=...,
    plugin_specific_config_b=...,
    ...
))
def foo(...) -> ...:
    ...

And

query_task = SnowflakeTask(
    query="Select * from x where x.time < {{.inputs.time}}",
    inputs=kwtypes(time=datetime),
    output_schema_type=pandas.DataFrame,
)

@workflow
def my_wf(t: datetime) -> ...:
    df = query_task(time=t)
    return process(df=df)

I'm trying to find examples/tutorials/how-tos on the first approach, but I can't find any. Can someone point me to an example?

Hey is there anyway to mock the NamedTuple output of a task and have that output then used in a seperate task? I have a situation where I have a task with some output

class GenerateOutputs(NamedTuple):
        uri: str
    
    @reference_task(
        ...
    )
    def generate(
        input:str 
    ) -> GenerateOutputs:
    ... 
    
    @workflow
    def my_wf(
        input="default_input"
    ) -> str:
    output = generate(
        input:str=input
    )
    return output.uri

However anytime I try to mock this out with something like

class TestNamedTuple(TestCase):
    def test_run_workflow(self):
        with task_mock(generate) as fake_generate:
            fake_generate.return_value = GenerateOutputs(uri="test_uri")
            uri = my_wf()
            self.assertIsNotNone(uri)

I get an error like :

AttributeError: 'Promise' object has no attribute 'uri'

How do I properly mock a task that has a namedTuple output with only one attribute?

hey flyte team, we finally got around to investigating migration to structured dataset! We have a few questions about the changes, maybe it makes sense to hop on a call? We're also open to submitting PRs if things make sense cc @Yee (who's on vaca 🙂 ) @Govind Raghu • There are some api structural changes, we're curious if they're intentional (eg requiring Annotated, loss of

open().write()

) • We have an issue setting the bigquery project for perms for bq URIs, there's a couple approaches to fixing this we can consider • There doesn't appear to be any type checking or casting of types based on the schema. this could be pretty valuable

https://github.com/flyteorg/flytekit/blob/master/setup.py#L49 any opposition to bumping

docker

to allow v6?

flytekit

is currently blocking us upgrading beyond v5.x

Does anyone know how to search through runs and outputs in python? I for some reason cannot figure out how to do that. I am trying to see how much metadata I can parse through in python. Thanks for your help!

hi team, i'm trying to fetch execution details similar to what's returned by

flytectl get execution execid --details --nodeID n0 -o yaml

execution = flyte_remote.fetch_execution(name = execution_id)
print (execution.node_executions)

after executing above

execution.node_executions

is not populated. I also tried

exec = flyte_remote.sync(execution)

but it returns the same

Hi, I’m excited to start using FlyteRemote. My first usecase is I’d like to retrieve a launchplan and modify a default input before executing. I have successfully done this for primitive inputs (str, int, etc.), but FlyteRemote complained when I tried to pass in a

@dataclass_json

class

Foo

. It said i needed

types.FooSchema

not

Foo

. I assume there is some way to transform

Foo

into

FooSchema

using either dataclass_json api or flyte api, but I’m not sure how to do it. Could you direct me? Exception and sample code in thread Thanks!

I couldn’t find this, but is there a way to specify the machine type in the task? I see options for cpu, gpu, and mem, but not for which types. I think we could train using AWS Sagemaker where the machine type in configurable, but seems unnecessary if I can just do it in Flyte. Let me know what I am missing! Thank you

Hi! I’m trying to sync an execution using

FlyteRemote.sync

but the response is too large:

grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
        status = StatusCode.RESOURCE_EXHAUSTED
        details = "Received message larger than max (4762259 vs. 4194304)"
        debug_error_string = "UNKNOWN:Error received from peer ipv4:192.168.3.75:81 {grpc_message:"Received message larger than max (4762259 vs. 4194304)", grpc_status:8, created_time:"2022-08-23T23:23:18.247266112+00:00"}"

Is there a way for me to adjust the max amount or somehow still get my synced execution data?

how is the description field in the protobuf populated for StructuredDatasets?

When registering a custom

TypeTransformer

w/ the

TypeEngine

, does the module need to be explicitly imported by one of the files containing workflows or tasks in order for it to be picked up? Based on the example of the

NumpyArrayTransformer

it seemed like it, but not sure where that needs to be declared

👋 is it possible to create an annotated

FlyteSchema

object that also includes metadata on whether a particular column is nullable vs required, along with its type?