Could it be that gate notes cannot be unblocked in conditionals? Video in the thread 🧵

AttributeError

when fetching inputs from

node_execution

for which the optional parameter of type

FlyteFile

is

None

from attr import s
import flytekit
import flytekit.remote
import flytekit.configuration as flyte_config
from typing import List, NamedTuple, Optional


import dataclasses, dataclasses_json

@dataclasses_json.dataclass_json
@dataclasses.dataclass
class T2IO:
    a: str
    b: str
    z: Optional[flytekit.types.file.FlyteFile] = None


@flytekit.task
def task1(x: int) -> NamedTuple('op', t1=List[T2IO]):
    return [T2IO(str(t), str(-t)) for t in range(x)]


@flytekit.task
def task2(x: T2IO) -> NamedTuple('op', t2=T2IO):
    return T2IO(x.a+'0', x.b+'0')


@flytekit.task
def task3(x: List[T2IO]) -> NamedTuple('op', t3=str):
    return '-'.join([t.a for t in x]+[t.b for t in x])


@flytekit.workflow
def baby_training_wf(x: int) -> str:
    t1, = task1(x=x).with_overrides(node_name='t-1')
    t2 = flytekit.map_task(task2)(x=t1).with_overrides(node_name='t-2')
    t3, = task3(x=t2).with_overrides(node_name='t-3')
    return t3

if __name__ == '__main__':
    print(baby_training_wf(x=3))

fetched execution and tried accessing task2 inputs as

execution.node_executions['t-2'].inputs['x']

, got

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
Cell In[308], line 1
----> 1 execution.node_executions['t-2'].inputs['x']

File /opt/homebrew/Caskroom/miniconda/base/envs/baby/lib/python3.10/site-packages/flytekit/core/type_engine.py:1691, in LiteralsResolver.__getitem__(self, key)
   1688 if key in self._native_values:
   1689     return self._native_values[key]
-> 1691 return self.get(key)

File /opt/homebrew/Caskroom/miniconda/base/envs/baby/lib/python3.10/site-packages/flytekit/core/type_engine.py:1720, in LiteralsResolver.get(self, attr, as_type)
   1718         else:
   1719             ValueError("as_type argument not supplied and Variable map not specified in LiteralsResolver")
-> 1720 val = TypeEngine.to_python_value(
   1721     self._ctx or FlyteContext.current_context(), self._literals[attr], cast(Type, as_type)
   1722 )
   1723 self._native_values[attr] = val
   1724 return val

File /opt/homebrew/Caskroom/miniconda/base/envs/baby/lib/python3.10/site-packages/flytekit/core/type_engine.py:831, in TypeEngine.to_python_value(cls, ctx, lv, expected_python_type)
    827 """
    828 Converts a Literal value with an expected python type into a python value.
    829 """
    830 transformer = cls.get_transformer(expected_python_type)
--> 831 return transformer.to_python_value(ctx, lv, expected_python_type)

File /opt/homebrew/Caskroom/miniconda/base/envs/baby/lib/python3.10/site-packages/flytekit/core/type_engine.py:1023, in ListTransformer.to_python_value(self, ctx, lv, expected_python_type)
   1021 else:
   1022     st = self.get_sub_type(expected_python_type)
-> 1023     return [TypeEngine.to_python_value(ctx, x, st) for x in lits]

File /opt/homebrew/Caskroom/miniconda/base/envs/baby/lib/python3.10/site-packages/flytekit/core/type_engine.py:1023, in <listcomp>(.0)
   1021 else:
   1022     st = self.get_sub_type(expected_python_type)
-> 1023     return [TypeEngine.to_python_value(ctx, x, st) for x in lits]

File /opt/homebrew/Caskroom/miniconda/base/envs/baby/lib/python3.10/site-packages/flytekit/core/type_engine.py:831, in TypeEngine.to_python_value(cls, ctx, lv, expected_python_type)
    827 """
    828 Converts a Literal value with an expected python type into a python value.
    829 """
    830 transformer = cls.get_transformer(expected_python_type)
--> 831 return transformer.to_python_value(ctx, lv, expected_python_type)

File /opt/homebrew/Caskroom/miniconda/base/envs/baby/lib/python3.10/site-packages/flytekit/core/type_engine.py:604, in DataclassTransformer.to_python_value(self, ctx, lv, expected_python_type)
    602 json_str = _json_format.MessageToJson(lv.scalar.generic)
    603 dc = cast(DataClassJsonMixin, expected_python_type).from_json(json_str)
--> 604 dc = self._fix_structured_dataset_type(expected_python_type, dc)
    605 return self._fix_dataclass_int(expected_python_type, self._deserialize_flyte_type(dc, expected_python_type))

File /opt/homebrew/Caskroom/miniconda/base/envs/baby/lib/python3.10/site-packages/flytekit/core/type_engine.py:392, in DataclassTransformer._fix_structured_dataset_type(self, python_type, python_val)
    390     for field in dataclasses.fields(python_type):
    391         val = python_val.__getattribute__(field.name)
--> 392         python_val.__setattr__(field.name, self._fix_structured_dataset_type(field.type, val))
    393 return python_val

File /opt/homebrew/Caskroom/miniconda/base/envs/baby/lib/python3.10/site-packages/flytekit/core/type_engine.py:391, in DataclassTransformer._fix_structured_dataset_type(self, python_type, python_val)
    389 elif dataclasses.is_dataclass(python_type):
    390     for field in dataclasses.fields(python_type):
--> 391         val = python_val.__getattribute__(field.name)
    392         python_val.__setattr__(field.name, self._fix_structured_dataset_type(field.type, val))
    393 return python_val

AttributeError: 'NoneType' object has no attribute 'path'

My papermill report is not getting added to the Flyte Deck. I am not sure if there is a setting on the cluster or I am missing something here. I can share more code if needed 🙂

quality_report = NotebookTask(
    name="quality_report",
    notebook_path=os.path.join(
        CURR_DIR, "demo_display.ipynb"
    ),
    render_deck=True,
    inputs=kwtypes(path=str),
    outputs=kwtypes(out_nb=PythonNotebook, out_rendered_nb=HTMLPage),
)

@workflow
def wf() -> Tuple[PythonNotebook, HTMLPage]:
    images = [score_image(name="img1.png"), score_image(name="img2.png"), score_image(name="img3.png")]
    display_grid(images=images)

    path = report_preprocessing(images=images)
    out, render = quality_report(path=path)
    return out, render

Hi I have a question regarding data flow in flyte - I read the docs on this and I understand that flyte only passes around reference to files, but if I have a large dataset in s3, for example, does that mean each task that has to process/use that data will download the entire dataset onto each task container’s memory?

Hi all, quick question: what is the correct way to pass a raw file or a pandas.DataFrame to a task as input using

FlyteRemote.execute

?

Hey team, I was wondering if anyone would be so kind as to show me how to pass a FlyteDirectory to a ContainerTask? There was a thread a few months ago that was related but I've been trying this from every angle and can't get the objects/files to show up in the container, here's how to reproduce:

bt = ContainerTask(
    name="basic-test",
    input_data_dir="/var/inputs",
    output_data_dir="/var/outputs",
    inputs=kwtypes(indir=FlyteDirectory),
    outputs=kwtypes(),
    image="<http://ghcr.io/flyteorg/rawcontainers-shell:v2|ghcr.io/flyteorg/rawcontainers-shell:v2>",
    command=[
        "ls",
        "-la",
        "/var/inputs",
    ],
)

@task
def get_dir(dirpath: str) -> FlyteDirectory:
    fd = FlyteDirectory(path=dirpath)
    return fd

@workflow
def wf():
    fd = get_dir(dirpath='<s3://my-s3-bucket/cv-in>')
    bt(indir=fd)

Running it with

pyflyte run --remote

, the above produces an empty

ls

output in the k8s logs. Trying to explicitly pass

/var/inputs/indir

to

ls

returns a "No such file or dir" exception for the task. Any help is appreciated as always!

I am experiencing a problem with caching in map tasks. For the following minimal example

import time

from typing import List

from flytekit import map_task, task, workflow

from flytekit.core.node_creation import create_node

USE_CACHE = True

CACHE_VERSION = "0.0.27"

SERIALIZE_CACHE = False

@task(

cache=USE_CACHE,

cache_version=CACHE_VERSION,

cache_serialize=SERIALIZE_CACHE,

)

def t1() -> str:

print("t1")

time.sleep(1)

return "t1"

@task(

cache=USE_CACHE,

cache_version=CACHE_VERSION,

cache_serialize=SERIALIZE_CACHE,

)

def t2(inp: str) -> str:

print("t2")

time.sleep(1)

return "t2"

@task(

cache=USE_CACHE,

cache_version=CACHE_VERSION,

cache_serialize=SERIALIZE_CACHE,

)

def t3_prep(inp: str) -> List[str]:

print("t3_prep")

return ["a", "b", "c"]

@task(

cache=USE_CACHE,

cache_version=CACHE_VERSION,

cache_serialize=SERIALIZE_CACHE,

)

def t3(inp: str) -> str:

print(f"t3 :: {inp}")

time.sleep(1)

return "t3"

@workflow

def wf1():

t1_out = t1()

t2_out = t2(inp=t1_out)

t3_prep_out = t3_prep(inp=t2_out)

t3_out = map_task(t3)(inp=t3_prep_out)

@workflow

def wf2():

wf1()

t1_out = t1()

t2_out = t2(inp=t1_out)

@workflow

def wf():

n1 = create_node(wf1)

n2 = create_node(wf2)

n1 >> n2

if __name__ == "__main__":

wf()

I get the (unexpected) output when running locally t1 t2 t3_prep t3 :: a t3 :: b t3 :: c t3 :: a t3 :: b t3 :: c Why is "t3" executed twice? I would expect it to be executed only once.

Hi everyone 🙂 I am trying to deploy Spark on Kubernetes, using the Dockerfile shown here : https://docs.flyte.org/projects/cookbook/en/latest/auto/integrations/kubernetes/k8s_spark/index.html#how-to-build-your-dock[…]or-spark-on-kubernetes I was trying to get rid of the VENV : is it necessary ? When I tried to remove it from the Dockerfile (I took care to make sure everything else worked), I had a bug when running a task.

Hi team.. is there an easy way to deactivate/archive all versions of a previously scheduled workflow? Or does that happen anyways when the latest is archived?

Hello, For slack notification to work, Flyte Admin needs to be configured just like email notification. However, I don’t see any explanation for configuring slack on the following doc. It only does for email. Could someone share the details for slack notification configs? Thanks! https://docs.flyte.org/en/latest/deployment/configuration/notifications.html

Where do I find RFCs for Flyte?

Hi, how can we change the default docker image for a

NotebookTask

? I have been using a

PodTemplate

to set the default docker image for a normal

@task()

, as per the link below. However the

NotebookTask

ignores the

image

setting in the

PodTemplate

, although it does apply other settings such as

VolumeMount

. How do I change the default docker image for a

NotebookTask

without using

pyflyte *--image xxxx run xxxx*

? https://docs.flyte.org/en/latest/deployment/configuration/general.html#using-default-k8s-podtemplates

Hi there! Is it possible to increase number of nodes in workflow? Is there a limit of 100 nodes with default settings? I have error from flytekit during graph registration:

"UNKNOWN:Error received from peer ipv6:%5B::1%5D:8089 {created_time:\"2023-04-25T17:31:47.560558+04:00\", grpc_status:3, grpc_message:\"number of nodes in workflow [{ResourceType:WORKFLOW Project:flytesnacks Domain:development Name:preprocessing Version:pBo5ZZ_OUYrcNTys_JOrZw== XXX_NoUnkeyedLiteral:{} XXX_unrecognized:[] XXX_sizecache:0}] exceeds limit (300 > 100)\"}\"\n>>"}

Is there a way to use a FlyteFile with an already uploaded path, in place of e.g. a

nn.Module

or

Tensor

when executing a workflow via

FlyteRemote

? The

FlyteFile[PYTORCH_FORMAT]

trick isn't working, though that may be something on my end. Should it? This works with

StructuredDataset

by setting the

uri

parameter as a remote path. So say I have a model uploaded at

<gs://my-bucket/my_location.pt>

and a workflows with

nn.Module

as an input parameter I'd like to run it with

flyte.execute(flyte.get_workflow(...), inputs={'model': FlyteFile(path='<gs://my-bucket/my_location.pt')})>

Or some such. Is this possible?

Hey everyone, I'm trying to run a wf on my AWS single binary deployment of Flyte. I've got port forwarding running with

kubectl -n flyte port-forward service/flyte-backend-flyte-binary 8088:8088 8089:8089

Config is this

admin:
  # For GRPC endpoints you might want to use dns:///flyte.myexample.com
  endpoint: localhost:8088
  authType: Pkce
  insecure: true
logger:
  show-source: true
  level: 0

And I'm getting this when running a wf:

$ FLYTE_CONFIG=~/.flyte/config.yaml pyflyte run --remote workflows/greeting.py wf --name Sam
Failed with Exception: Reason: SYSTEM:Unknown
RPC Failed, with Status: StatusCode.UNAVAILABLE
	details: failed to connect to all addresses; last error: INTERNAL: ipv4:127.0.0.1:8088: Trying to connect an http1.x server
	Debug string UNKNOWN:failed to connect to all addresses; last error: INTERNAL: ipv4:127.0.0.1:8088: Trying to connect an http1.x server {grpc_status:14, created_time:"2023-04-25T15:31:54.782127+01:00"}

Hello 🙂, first things first: Thanks for all your work! Flyte is such a great tool! ❤️ Right now we try to establish Flyte in our company, however before we use it in production there are some obstacles to be removed. So my first question: How can I inject secret strings into the backend configuration e.g. S3/MinIO secret keys (We installed our Flyte backend with Helm and I provide a values.conf)

provider: s3
    providerConfig:
      s3:
        disableSSL: true
        v2Signing: true
        endpoint: <http://XXXXXXXXXX:9000>
        authType: accesskey
        accessKey: XXXXX <<<<-------
        secretKey: XXXXXXXXXXXXXXX <<<<---

Of course I cant write it into the config file directly. Is there a straightforward way I dont see here? Like Env Vars? Thanks in advance!

Hi flyte community! I have a question about the resolver path. I noticed that if I use

pyflyte register

to register the workflow, the resolver path would be

app.<…>.default_task_resolver

, but if programmatically register workflow using

register_workflow()

, the resolver path would be

module.<…>.default_task_resolver

(

module

being the root of the workflow code). Is there a way to change the resolver path back to

app

if I use

register_workflow

?

Hi again everyone, @David Espejo (he/him) has been helping me with my AWS deployment of Flyte. We're nearly there, got a wf deployed, but it just hangs and doesn't complete or fail. I'm trying to execute the greeting wf (works fine in the sandbox). These are the events from the

kubectl describe pods

cmd. Anyone got any ideas?

Type  Reason   Age  From        Message
 ----  ------   ----  ----        -------
 Normal Scheduled 8m56s default-scheduler Successfully assigned flyte/flyte-backend-flyte-binary-5b978865bb-r757k to ip-10-250-3-158.eu-west-1.compute.internal
 Normal Pulled   8m56s kubelet      Container image "postgres:15-alpine" already present on machine
 Normal Created  8m56s kubelet      Created container wait-for-db
 Normal Started  8m55s kubelet      Started container wait-for-db
 Normal Pulled   8m55s kubelet      Container image "<http://cr.flyte.org/flyteorg/flyte-binary-release:v1.4.3|cr.flyte.org/flyteorg/flyte-binary-release:v1.4.3>" already present on machine
 Normal Created  8m55s kubelet      Created container flyte
 Normal Started  8m54s kubelet      Started container flyte

The

kubectl log

cmd doesn't contain anything useful, just a bunch of SQL commands

Hi guys! Sometimes we receive error:

code:"ContainersNotReady|CreateContainerConfigError" message:"containers with unready status: [amxj9hwsgs88pkjn5mxc-n3-0-n5-0-43]|failed to sync secret cache: timed out waiting for the condition"

We understand that this is the issue of our Kubernetes setup, but why does Flyte not retry this type of error? After the first occurrence, Flyte labels this task as failed.

Hi everyone, Is it possible to run part of a worklow or a list of tasks in a single pod? I would like to use the caching mechanism for tasks but reduce the cost of starting a pod for each task. I have a task that i want to call up thousands of times, but it doesn't require a lot of computing time

Hello, While registering a workflow I got this weird error. It seems all of the tasks are complaining about the same. Could you provide any idea why this might be happening?

grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
	status = StatusCode.INVALID_ARGUMENT
	details = "task with different structure already exists with id resource_type:TASK project:"marketing-ds" domain:"development" name:"dai_mle_models.xgboost.shared_tasks.preprocess" version:"13ae7115d121144d807168e49caaa0aaa25faa14" "
	debug_error_string = "UNKNOWN:Error received from peer ipv4:3.82.167.33:443 {grpc_message:"task with different structure already exists with id resource_type:TASK project:\"marketing-ds\" domain:\"development\" name:\"dai_mle_models.xgboost.shared_tasks.preprocess\" version:\"13ae7115d121144d807168e49caaa0aaa25faa14\" ", grpc_status:3, created_time:"2023-04-26T11:06:59.
...
During handling of the above exception, another exception occurred:
...
File "/Users/fshen/hbo-code/dai-mle-subltv/env/lib/python3.8/site-packages/flytekit/clients/raw.py", line 82, in handler
    e.details += "create_request: " + _MessageToJson(create_request)
TypeError: unsupported operand type(s) for +=: 'method' and 'str'

Hi team.. Quick question about schedules. We have a workflow that has a launch plan for it to run once every 8 hours. The workflow does not take any inputs and tasks are not cached. For some bizarre reason it has been firing twice with the same version at the same instance for several days now. Any idea what could cause this? Here is the launch plan

{
  "id": {
    "resourceType": "LAUNCH_PLAN",
    "project": "my_project",
    "domain": "production",
    "name": "my_lp",
    "version": "20949aa7bbea5421dfef9f96d4248c7a26a6f5d5"
  },
  "spec": {
    "workflowId": {
      "resourceType": "WORKFLOW",
      "project": "my_project",
      "domain": "production",
      "name": "some_wf",
      "version": "20949aa7bbea5421dfef9f96d4248c7a26a6f5d5"
    },
    "entityMetadata": {
      "schedule": {
        "rate": {
          "value": 8,
          "unit": "HOUR"
        }
      }
    },
    "defaultInputs": {},
    "fixedInputs": {},
    "labels": {},
    "annotations": {},
    "rawOutputDataConfig": {}
  },
  "closure": {
    "state": "ACTIVE",
    "expectedInputs": {},
    "expectedOutputs": {},
    "createdAt": "2023-04-26T18:23:39.503710Z",
    "updatedAt": "2023-04-26T18:23:39.503710Z"
  }
}

Is there a way to prevent concurrent execution of workflows or tasks? I see https://github.com/flyteorg/flyte/issues/420 but no traction. For tasks there's

cache_serialize

but this only works if caching is enabled and for tasks with the same input - I want to serialize all tasks of the same type. Any suggestions?

Hello all Can someone help me with a dynamic workflow situation? In local tests it runs OK, but when I upload it to Flyte in my cluster, it queue the tasks but never run them... The picture shows what's happening.

Hello 👋 community! Appreciate your great work! I have a potentially quick question, can the flyte serialization settings take a image digest (with

@sha256

) in anyway for example:

image_config = ImageConfig(default_image=Image(name=container_image_url))

    ss = SerializationSettings(
        image_config=image_config,
        project=execution_options.project,
        domain=execution_options.domain,
        version=version,
    )

where

contaienr_image_url

is an image digest, and it would give me such error

TypeError: __init__() missing 2 required positional arguments: 'fqn' and 'tag'

I checked the source code and it seems like the image full path can only be name-tag but not digest?

class Image(object):
    """
    Image is a structured wrapper for task container images used in object serialization.

    Attributes:
        name (str): A user-provided name to identify this image.
        fqn (str): Fully qualified image name. This consists of
            #. a registry location
            #. a username
            #. a repository name
            For example: `hostname/username/reponame`
        tag (str): Optional tag used to specify which version of an image to pull
    """

    name: str
    fqn: str
    tag: str

    @property
    def full(self) -> str:
        """ "
        Return the full image name with tag.
        """
        return f"{self.fqn}:{self.tag}"

Hi! It’s me again. Sorry for piling on the questions. I wonder what’s the python equivalence of this

Retrieve all the workflows with limit and sorting:

flytectl get -p flytesnacks -d development workflow  --filter.sortBy=created_at --filter.limit=1 --filter.asc

Thanks!

Hi everyone! I recently started working with Flyte and have a workflow with conditional branching that kicks of one of two other sub workflows. I need a way to programmatically fetch from python if a specific task within an execution of that workflow was run and if it was if it finished successfully. Even looking into the flytekit code, I can't find a simple way to do this. Ideally I don't need to get the state of the entire workflow, just a single task. The branching logic also complicates it from what I've noticed. I was able to do this in a static workflow by pulling out some code from the

sync_execution()

method. Maybe I'm missing something?

RESOLVED! Hey guys, i have a problem with a very little task at the moment:

@task(
    timeout=timedelta(hours=1),
    requests=Resources(
        cpu="100m",
        mem="300Mi"
    ),
)
def prepare_list(website_id_first: int, website_id_last: int) -> typing.List[int]:
    if website_id_first > website_id_last:
        raise ValueError(
            "website_id_last has to be larger than website_id_first")
    l: typing.List[int] = list(range(website_id_first, website_id_last+1))
    test: typing.List[int] = [1, 2, 3]
    print("l is type " + str(type(l)) + "and has content " + str(l))
    return test

No matter what im doing with it i always get the Error: "Outputs not generated by task execution" Any ideas what i am doing wrong?

A question about deck. When I click the

flyte deck

button in the console, instead of rendering the deck in browser, it downloads a html file which has the deck content. Any one knows how to solve the issue?

I'm working with a deployed version of flyte in k8s that leveraged the contour reverse proxy. I also have istio service mesh deployed. This causes issues with communication. Anyone know what specific config changes might be needed to get istio to work with the contour reverse proxy? For background, I don't need flyte available outside the cluster so an ingress point isn't required.