Hello Flyte Team! I want to use custom built docker image to use our custom dependencies. I tested on flyte sandbox on my local macbook. (Running from

flytectl demo start

) My workflow was successfully run on local (

pyflyte run cifar10.py wf

) But my workflow needs

torch

packages, so I built docker image with Dockerfile like below:

FROM python:3.9-slim-buster

RUN python -m pip install torch torchvision

RUN python -m pip install flytekit pandas numpy awscli

I pushed the image to dockerhub, it can be accessed publicly. And I submit flyte workflow with

pyflyte

cli

pyflyte run --remote --image {username}/flyte:0.0.4 cifar10.py wf

But it doesn’t work on remote (sandbox). How can I do?

Hi is it possible to have a

task

fail but still continue the

workflow

? Our use case is a

dynamic

workflow spawns N

tasks

, and it’s ok if some small percentage fail. We’re currently using a try catch to return an empty result (likely could also use an

Optional

), but then the user loses visibility by clearly seeing the

task

as failed in the console.

Hey, How does your development process in Flyte look like? I mean, do you manage a separation between environments - dev and prod for example? If so, how? Thanks everyone!

Hi! I have a few questions about Flyte caching.. I read through the documentation here a few times but I’m still a bit confused about why the “Caching of Non-flyte Offloaded Objects” examples are hashing the task output instead of the input. My intuition for how a cache would work is that the input values for a task needs to be hashable so that they can be used as keys for the cache. In which case, I feel like it would make more sense to add the

Annotated HashMethod

to the input types of a task instead of the output. Then, the

HashMethod

can be run on the inputs when a cached task is invoked to determine the cache keys and then check if there's a hit. I tried this out and it kind of works for local caching? but not for remote.

@task(cache=True, cache_version="1.0")
def my_task(obj: ty.Annotated[pd.DataFrame, HashMethod(hash_df)]) -> pd.DataFrame:
    redis_client.incr(REDIS_COUNTER_KEY)

    return obj

@dynamic
def my_workflow():
    obj = pd.DataFrame(
        {
            "name": ["a", "b"],
            "val": ["test1", "test2"],
        }
    )

    obj = my_task(obj=obj)
    obj = my_task(obj=obj)
    my_task(obj=obj)

In the example above

my_task

will be be called three times the first time

my_workflow

is called. This still doesn't match my expectation since I thought it would be called once for the first call and for the cache to be hit on the second and third call since the input

obj

is the same. However! The second run of

my_workflow

has a cache hit for all three calls to

my_task

so it does work to some extent even though I don’t fully understand what it’s doing. For remote caching, this doesn’t seem to work at all and there are no cache hits no matter how many times I run

my_workflow

.

Is it possible to cancel/terminate individual tasks in a workflow somehow?

I've spun up a Sandbox cluster (using

flytectl sandbox start

) on one machine, and I'm trying to run an example wf from different machine (I've updated

endpoint

in

~/.flyte/config.yaml

).

pyflyte run --remote example.py:wf --n 500 --mean 42 --sigma 21

succeds and I get a link to the console. However when inspecting the execution I see this error:

[1/1] currentAttempt done. Last Error: USER::Pod failed. No message received from kubernetes.
[f31016626d79e40b2a4c-n0-0] terminated with exit code (1). Reason [Error]. Message: 
_execute(cmd=cmd, s3_cfg=self.s3_cfg)
  File "/usr/local/lib/python3.8/site-packages/flytekit/extras/persistence/s3_awscli.py", line 51, in _update_cmd_config_and_execute
    return subprocess.check_call(anonymous_cmd, env=env)
  File "/usr/local/lib/python3.8/site-packages/flytekit/tools/subprocess.py", line 26, in check_call
    raise Exception(
Exception: Called process exited with error code: 1.  Stderr dump:

b'fatal error: An error occurred (403) when calling the HeadObject operation: Forbidden\n'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/bin/pyflyte-fast-execute", line 8, in <module>
    sys.exit(fast_execute_task_cmd())
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1130, in __call__
    return self.main(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1055, in main
    rv = self.invoke(ctx)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1404, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 760, in invoke
    return __callback(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/flytekit/bin/entrypoint.py", line 495, in fast_execute_task_cmd
    _download_distribution(additional_distribution, dest_dir)
  File "/usr/local/lib/python3.8/site-packages/flytekit/tools/fast_registration.py", line 102, in download_distribution
    file_access.get_data(additional_distribution, destination)
  File "/usr/local/lib/python3.8/site-packages/flytekit/core/data_persistence.py", line 427, in get_data
    raise FlyteAssertion(
flytekit.exceptions.user.FlyteAssertion: Failed to get data from <s3://my-s3-bucket/36/flytesnacks/development/VGJSBXLCAE3JIKQYKY2O3N5IMQ======/scriptmode.tar.gz> to . (recursive=False).

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

b'fatal error: An error occurred (403) when calling the HeadObject operation: Forbidden\n'
.

Hi there! I'm interested in the ability to run flyte workflows with priorities so that workflows with a higher priority are executed before those with a lower one when they are running at the same time. I found some information about this in the documentation but did not understand how exactly you can start the workflow by specifying its priority as a parameter. Another question is - how can resources be managed? Let's say I have a CPU-intensive workflow, and it was experimentally revealed that my cluster is capable of processing up to 10 such workflows at the same time, what will happen if we run 20 workflows?

Hey, is there documentation for the new pod templates feature?

Hi people! I have not found a lot of documentation about the Prometheus metrics made available by Flyte. Are there metrics about tasks such as individual task completion times, durations, etc? Much like we have for workflows with metrics like "flytepropellerallworkflowsuccess_duration_ms_count". These informations are on Flyte console so I wonder if they are also available by metrics or if at least there is a plan to implement that. Also, are there metrics about CPU and memory usage by workflow/task?

Hi team... quick question. If a task calls a task, do they both not show up in the Flyte console? I only see the outer task