Hi, can you push a folder to minio server within flyte cluster setup with single binary and use it in tasks by downloading from minio server. Any suggestion

Hi everyone, I'm having trouble configuring UI logging links for flyte tasks that use the ray plugin. I can see logging links being rendered and working just fine for normal flyte tasks, but I don't see any logging links being rendered for

Ray

tasks. Do I need to configure something separately from the template URIs in the propeller configmap for ray tasks to give me a link? I couldn't find anything clear in the docs or previous slack messages unfortunately. Thanks, and appreciate your help!

Hi, I have tasks that connects minio and download . This works perfectly fine when run local Same one fails when running in remote. Note that client.list_objects works on remote(pyflyte run --remote), but fails on iterating the object list( `for obj in objects`:)

Message:

    MaxRetryError: HTTPConnectionPool(host='localhost', port=9000): Max retries exceeded with url: /flyte?location= (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x79443cd44d70>: Failed to establish a new connection: [Errno 111] Connection refused'))

@task(container_image=image_spec)
def download_wod_ds()->FlyteDirectory:
# Set up Minio client
    client = Minio(
        "localhost:9000",
        access_key="BBsho4hEODTw0J2sv6uk",
        secret_key="rQn3WKDAhci6EnfwJf48XjpIpkbXInx0R1ciMNG7",
        secure=False
    )

    # Specify the bucket and folder to download
    bucket_name = "flyte"
    folder_name = "ns-wod-ds"

    # Create the download directory
    base_path = os.getcwd()
    new_directory_path = os.path.join(base_path, "wod-ds")
    os.makedirs(new_directory_path, exist_ok=True)
    print("made bucket")
    # List all objects in the folder
    objects = client.list_objects(bucket_name, prefix="flytesnacks/development/ns-wod-ds/training", recursive=True)
    print("ran list_objects")
    # Download each object to the local directory. FAIL'ed from here when run remote.
    for obj in objects:
        print("going thru object ist")
        object_name = obj.object_name
        download_path = os.path.join(new_directory_path, os.path.relpath(object_name, folder_name))
        download_dir_path = os.path.dirname(download_path)
        os.makedirs(download_dir_path, exist_ok=True)
        print("before downloading")
        try:
            client.fget_object(bucket_name, object_name, download_path)
            print(f"Downloaded {object_name} to {download_path}")
        except S3Error as exc:
            print(f"Error downloading {object_name}: {exc}")
    return FlyteDirectory(path=new_directory_path)

I have setup flyte using flyte-binary single node So, 1. what is wrong in my code 2. Is there anyway to do portforwarding in PodTemplate if Connectivity is an issue.

I have a very subtle bug. My workflow looks similar to this:

my_image = ImageSpec(name='my_custom_image',
                     builder='noop',
                     base_image='my-custom/base-image:abc')

@task(container_image=my_image)
def create_dataframe() -> str:
    return "DataFrame created successfully!"

@dynamic
def my_dynamic_workflow() -> str:
    """
    This dynamic workflow calls the task.
    """
    # The dynamic workflow context automatically runs this task.
    result = create_dataframe()
    return result

When then

@dynamic

workflow runs, it tries to kick off a

task

with the default flyte image instead of my

my-custom/base-image:abc

image. Now clearly the image builder is not supposed to run during a Flyte execution itself (at least i see an if-guard to that effect)... but this is the no-op builder which doesn't actually need to build. So what inside the

dynamic

workflow is preventing my `ImageSpec`'s image name from NOT getting reported properly?

Hi all I've been trying to do a small experiment with flyte sandbox (sha-08a8d43838a8c02b07f18394ed74d5d17ce61891), however I keep seeing the error that says:

TypeTransformerFailedError: Error converting input 'df' at position 0:
    Literal value: Flyte Serialized object (Literal):
      scalar:
        structured_dataset:
          uri: <s3://my-s3-bucket/data/cn/arnr4n65vn287zp446f5-n2-> [...]
          metadata:
            structured_dataset_type:
              format: parquet
    Expected Python type: <class 'pandas.core.frame.DataFrame'>
    Exception: Expected checksum ++koqw== did not match calculated checksum: hniEMQ==

It seems to be happening when the pandas dataframe gets transferred between tasks. Here is the simplified workflow:

import os

import flytekit as fl
import pandas as pd
import numpy as np

from datetime import datetime, timedelta

from src.util.logging import get_logger, setup_logging

logger = get_logger(__name__)

CACHE_VERSION = "0.1"

image_spec = fl.ImageSpec(
    name="weather-app",
    requirements="uv.lock",
    registry=os.environ["FLYTE_IMAGE_REGISTRY"],
)

request_resources = fl.Resources(cpu="1", mem="1000Mi", ephemeral_storage="500Mi")
limit_resources = fl.Resources(cpu="2", mem="2000Mi", ephemeral_storage="1000Mi")


@fl.task(container_image=image_spec)
def round_datetime_to_hour(dt: datetime) -> datetime:
    return datetime(year=dt.year, month=dt.month, day=dt.day, hour=dt.hour, tzinfo=None)


@fl.task(container_image=image_spec)
def get_data(start: datetime, end: datetime) -> pd.DataFrame:
    n_rows = 1_000_000
    # n_rows = 1_000
    <http://logger.info|logger.info>(f"running get_data with start={start}, end={end}")

    np.random.seed(42)

    start_date = datetime(2020, 1, 1)
    end_date = datetime(2024, 12, 31)
    dates = pd.date_range(start=start_date, end=end_date, periods=n_rows)

    df = pd.DataFrame(
        {
            "date": dates.astype("datetime64[s]"),
            "price": np.random.uniform(10.0, 1000.0, n_rows),
            "quantity": np.random.randint(1, 1000, n_rows, int),
        }
    )

    <http://logger.info|logger.info>(f"df dtypes={df.dtypes}")

    return df


@fl.task(container_image=image_spec)
def log_data(df: pd.DataFrame):
    <http://logger.info|logger.info>(f"data={df.head()}")


@fl.workflow
def experiment(
    start: datetime = datetime.now().replace(
        tzinfo=None, hour=0, minute=0, second=0, microsecond=0
    )
    - timedelta(days=3),
    end: datetime = datetime.now().replace(
        tzinfo=None, hour=0, minute=0, second=0, microsecond=0
    ),
):
    start_date = round_datetime_to_hour(start)
    end_date = round_datetime_to_hour(end)
    df = get_data(start_date, end_date)
    log_data(df)


setup_logging(level="INFO")

I noticed that if the number of rows is 1000 the error does not happen(n_rows variable). Looks like something related to serialisation, so I tried to store it as StructuredDataset, however that did not help. Any pointers on what can be the problem and how to fix it?

Hi, I've tried to deploy flyte(-core) into AWS infrastructure (EKS) with https://github.com/unionai-oss/deploy-flyte ... I do not have ingress properly set up, so I'd like to rely on port-forwardning. However I have no idea which service to port-forward for console (is it svc/flyteconsole? That only shows empty project page and no other info) and what should I port-forward for my ~/.flyte/config.yaml endpoint... I have tried

kubectl port-forward -n flyte svc/flyteadmin 30080:81

, now

pyflyte info

actually shows some information so it seems like it connects to something. But when I run the code example from https://github.com/unionai-oss/deploy-flyte/blob/main/environments/aws/flyte-core/README.md using

pyflyte run --remote hello_world.py my_wf

it tells me Running Execution on Remote. 00000 Running execution on remote. [✔️] Go to http://localhost:30080/console/projects/flytesnacks/domains/development/executions/afr2zbgj2bkp5lfrv9sr to see execution in the console. Of course there's nothing on that page, and my localhost:8080 (where I have svc/flyteconsole:

kubectl port-forward -n flyte svc/flyteconsole 8080:80

) does not show any project info either

im having some trouble with map_task and would appreciate support my task:

def *train_aggregator*(_fold_: str) -> str:

in workflow:

map_task(train_aggregator)(_fold_=['fold1', 'fold2'])

but error is complaining about mismatch between string and list of strings

he output variable 'main.map_train_aggregator_6b3bd0353da5de6e84d7982921ead2b3-arraynode.o0' has type [collection_type:{simple:STRING}], but it's assigned to the input variable 'o0' which has type type [simple:STRING].

from my understanding, and confirmed by LLM, map_task should handle this

One question I have, I was not really able to make it work so far. How do you return a FlyteDirectory from a ContainerTask ? The documentation has exemple with FlyteFiles, and it does not seems to work in my exemple code below:

from flytekit import ContainerTask, kwtypes, FlyteDirectory

produce_data = ContainerTask(
    name="produce_data",
    image="alpine:3.18",
    command=["sh", "-c"],
    arguments=[
        # Important: write to the output path that Flyte provides
        "mkdir -p /var/flyte/output/results && echo 'hello' > /var/flyte/output/results/hello.txt"
    ],
    output_data_dir="/var/flyte/output",  # :white_check_mark: ensures Flyte watches this dir
    outputs=kwtypes(results=FlyteDirectory),
)

Do you guys have some working exemple of ContainerTask returing FlyteDirectory ?

👋 is there any way to configure a higher default for a task's

retries

parameter? we'd like to modify an internal method to use

FlyteRecoverableException

but we need a way to bump the default

retries

on all tasks. I noticed a default-max-attempts parameter on flyte propeller, but not entirely sure if this translates to task's

retries

parameter

Has anyone tried this

enableArtifacts

feature in Flyte? Curious if folks have had any trouble using it? https://www.union.ai/docs/v1/flyte/deployment/configuration-reference/flyteadmin-config/#featuregates-interfacesfeaturegates

Hey, I'm developing a new workflow where one of the outputs of a task is a

torch.Tensor

that is an input to another task. I notice that when regardless if the tensor is on the GPU or CPU when it's outputted, it's put back on the GPU in the next task when running locally. This is a bit unexepected behaviour, as I would expect it deserialize/load in the next step on the device it was serialized on (i.e. if the tensor was on CPU when outputted, I expect it to be loaded as a CPU tensor. If the tensor is on GPU, I expect it to be loaded on GPU tensor). Running flytekit==1.15.3 (locally)

If this is solved in a newer version of flyte v1, happy to update

Hi, I'm using flytekit 1.14 and I'm trying to use map tasks and dynamic tasks but I keep getting

ModuleNotFoundError: No module named 'site-packages'

This is the complete traceback:

/opt/venv/bin/pyflyte-execute:8 in <module>                                  │
│                                                                              │
│ ❱ 8 │   sys.exit(execute_task_cmd())                                         │
│                                                                              │
│ /opt/venv/lib/python3.10/site-packages/click/core.py:1161 in __call__        │
│                                                                              │
│ ❱ 1161 │   │   return self.main(*args, **kwargs)                             │
│                                                                              │
│ /opt/venv/lib/python3.10/site-packages/click/core.py:1082 in main            │
│                                                                              │
│ ❱ 1082 │   │   │   │   │   rv = self.invoke(ctx)                             │
│                                                                              │
│ /opt/venv/lib/python3.10/site-packages/click/core.py:1443 in invoke          │
│                                                                              │
│ ❱ 1443 │   │   │   return ctx.invoke(self.callback, **ctx.params)            │
│                                                                              │
│ /opt/venv/lib/python3.10/site-packages/click/core.py:788 in invoke           │
│                                                                              │
│ ❱  788 │   │   │   │   return __callback(*args, **kwargs)                    │
│                                                                              │
│ /opt/venv/lib/python3.10/site-packages/flytekit/bin/entrypoint.py:715 in     │
│ execute_task_cmd                                                             │
│                                                                              │
│ ❱ 715 │   _execute_task(                                                     │
│                                                                              │
│ /opt/venv/lib/python3.10/site-packages/flytekit/bin/entrypoint.py:579 in     │
│ _execute_task                                                                │
│                                                                              │
│ ❱ 579 │   │   resolver_obj = load_object_from_module(resolver)               │
│                                                                              │
│ /opt/venv/lib/python3.10/site-packages/flytekit/tools/module_loader.py:51 in │
│ load_object_from_module                                                      │
│                                                                              │
│ ❱ 51 │   class_obj_mod = importlib.import_module(".".join(class_obj_mod))    │
│                                                                              │
│ /usr/lib/python3.10/importlib/__init__.py:126 in import_module               │
│                                                                              │
│ ❱ 126 │   return _bootstrap._gcd_import(name[level:], package, level)        │
│ in _gcd_import:1050                                                          │
│ in _find_and_load:1027                                                       │
│ in _find_and_load_unlocked:992                                               │
│ in _call_with_frames_removed:241                                             │
│ in _gcd_import:1050                                                          │
│ in _find_and_load:1027                                                       │
│ in _find_and_load_unlocked:992                                               │
│ in _call_with_frames_removed:241                                             │
│ in _gcd_import:1050                                                          │
│ in _find_and_load:1027                                                       │
│ in _find_and_load_unlocked:992                                               │
│ in _call_with_frames_removed:241                                             │
│ in _gcd_import:1050                                                          │
│ in _find_and_load:1027                                                       │
│ in _find_and_load_unlocked:1004                                              │
╰──────────────────────────────────────────────────────────────────────────────╯
ModuleNotFoundError: No module named 'site-packages'

I tried using @ancient-wolf-19325 and it says to use python project structure which I did but I still keep getting either the same

ModuleNotFoundError

or

Empty module name

I'm 99% sure I did something stupid when configuring the helm chart to enable pytorch, but I keep getting the following error:

pytorchjobs.kubeflow.org is forbidden: User "system:serviceaccount:flyte:flyte-backend-flyte-binary" cannot create resource "pytorchjobs" in API group "kubeflow.org" in the namespace "inference-staging"

Relevant section of the helm chart in thread

Does the task timeout include any time the task might be queued by k8s due to there not being enough CPUs / GPUs available? Or does the clock start ticking only when Flyte gets to the "Execute user level code." stage in the container?

Flyte Team, when will Flyte 1.16 be released? 🙂

Hi, could you help me with AWS Batch job setup? • I have flyte deployed into AWS/EKS using deploy-flyte TF • I updated my configuration with https://www.union.ai/docs/v1/flyte/deployment/flyte-plugins/batch/ so that the AWS batch plugin is configured in the chart (I'm not sure if I entered all values correctly, since the documentaiton is not very descriptive). For example, what should be roleNameKey value here? I added the value of flyte_backend_irsa_role that was set up with deploy-flyte • the confuguration for flyteadmin and flytepropeller from the page above already includes job queue names that should be used • Now I'm trying the example of https://www.union.ai/docs/v1/flyte/integrations/external-service-backend-plugins/aws-batch-plugin/batch/ Where there is only AWSBatch configuration added as task_config (I did not add any parameters nor platformCapabilities as I was hoping some good defaults would be used Now I registered the flow in flyte and I am told I should execute it from UI. UI asks me for Role ARN and Kubernetes service account. Do I need to provide these when I added them to the server configuration in the plugin setup?

HI Folks, I am wondering if there is a good way to rename a task? I have a few tasks (all decorated with

@task

) but I can't find a good way to rename them. We use the task name for some of our metrics but they are too long and I would like to shorten them

Hi - is there a channel to discuss features of flyte 2? i couldn't find a reference in the docs to a flyte-native mechanism for scheduling a task

Howdy - I'm playing around with the new

WebhookTask

. i got it to work but i need to explicitly add

httpx

to the imagespec. is that expected? i assumed a feature like would be encapsulated in the base image (

@task(container_image="<http://ghcr.io/flyteorg/flytekit:py3.12-latest|ghcr.io/flyteorg/flytekit:py3.12-latest>")

). i'm kinda a python noob, so apologies if this is an obvious package management question

Hi folks.. is there any plan to address this? We are dealing with a growing cache as well and would like to set a lifecycle and clean up without the fear of breaking any reruns.

Will flyte 1.16 be released? I see that the last beta version is from end of july, is it stable? Or are all developments for flyte 1 on hold/discontinued in favor of flyte 2?

Hey Team, we would like to test our Flyte setup with databricks agent but we are blocked due to following issue, just wanted to know if there any plan to fix and release it soon. https://github.com/flyteorg/flyte/issues/6465 @nutritious-whale-39926@mysterious-country-97979 /@helpful-mechanic-41804

this doesn't seem easily done, but just in case i'm missing something; is there a way to identify the total cpu/gpu hours of completed tasks in v1? ie some way to pull the podspec given the execution using flytekit?

docker run --rm -it <http://cr.flyte.org/flyteorg/datacatalog-release:v1.16.0|cr.flyte.org/flyteorg/datacatalog-release:v1.16.0> /bin/sh
/ $ vi /tmp/otel.yaml
/ $ cat /tmp/otel.yaml
otel:
  file: /tmp/trace.txt
  jaeger:
    endpoint: <http://localhost:14268/api/traces>
  otlpgrpc:
    endpoint: <http://localhost:4317>
  otlphttp:
    endpoint: <http://localhost:4318/v1/traces>
  sampler:
    parentSampler: always
  type: noop
/ $ datacatalog --config /tmp/otel.yaml migrate run
INFO[0000] Using config file: [/tmp/otel.yaml]
Error:

1 error(s) decoding:

* 'file' expected a map, got 'string'

Looks like at a glance the new otel config in the helm chart doesn't match what the various components are expecting.

I have a repo with the following structure

.
├── local_scripts
│   ├── program1_input.csv
│   └── program1_quickrun.py
├── pyproject.toml
├── README.md
└── src
    └── internal_flytekit
        ├── __init__.py
        ├── tasks
        │   ├── __init__.py
        │   └── program1.py
        ├── utils.py
        └── workflows
            ├── __init__.py
            └── program1.py

I'm running short test scripts in

local_scripts

via

python local_scripts/program1_quickrun.py

The contents of

local_scripts/program1_quickrun.py

are very much like:

#!/usr/bin/env python

import os

FLYTE_TASK_VERSION = "17.4.0-alpha.2.program1"
os.environ["JNJ_FLYTE_IMAGE_VERSION"] = FLYTE_TASK_VERSION

from flytekit import Config, FlyteFile, FlyteRemote

from internal_flytekit.workflows.program1 import (
    run_program1_wf,
)


def main() -> None:
    remote = FlyteRemote(
        config=Config.for_endpoint(endpoint="<http://flyteserver.internal.com|flyteserver.internal.com>"),
        default_project="flyteproject",
        default_domain="predev",
        interactive_mode_enabled=True,
    )
    exe = remote.execute(
        run_program1_wf,
        inputs={
            "input_csv": FlyteFile("local_scripts/program1_input.csv"),
        },
    )
    print("job", exe.execution_url)

if __name__ == "__main__":
    main()

When i run this workflow directly via pyflyte run: it works fine because it's packaging the relevant parts of internal_flytekit in a tar.gz file and serving that to the worker. but when i just run

python local_scripts/program1_quickrun.py

it is only packaging a pkl file and the task fails with the error:

Exception when loading_task, reason USER:RuntimeError: error=No module named 'internal_flytekit', cause=No module named 'internal_flytekit'

Is there something that I'm doing wrong here? or some way to convince the python script to package the whole repo or the relevant parts of the repo with the task?

Hi folks… I’m seeing this error when I try to launch a task. Has anyone else run into this?

Hi everyone! 👋 I'm trying to access the Ray cluster dashboard via the Flyte UI. I was wondering if there's a recommended way to enable or expose the Ray dashboard link directly in the Flyte UI when running Ray tasks? If anyone has done this before or has pointers on how to configure it (either via Flyte backend or FlyteRay plugin settings), I'd really appreciate your help!

Hey guys!! I've got a question we've been trying to solve for a while. We're packaging a module containing a workflow like this:

pyflyte --pkgs my_workflows package -f

Everything works fine, then we register it:

flytectl register files \

--project $PROJECT \

--domain $DOMAIN \

--archive $ARCHIVE \

--version "$VERSION"

and when we run it we see the following error:

Trace:

    Traceback (most recent call last):
      File "/usr/local/lib/python3.12/site-packages/flytekit/bin/entrypoint.py", line 182, in _dispatch_execute
        task_def = load_task()
                   ^^^^^^^^^^^
      File "/usr/local/lib/python3.12/site-packages/flytekit/bin/entrypoint.py", line 604, in load_task
        return resolver_obj.load_task(loader_args=resolver_args)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      File "/usr/local/lib/python3.12/site-packages/flytekit/core/utils.py", line 309, in wrapper
        return func(*args, **kwargs)
               ^^^^^^^^^^^^^^^^^^^^^
      File "/usr/local/lib/python3.12/site-packages/flytekit/core/python_auto_container.py", line 302, in load_task
        task_module = importlib.import_module(name=task_module)  # type: ignore
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      File "/usr/local/lib/python3.12/importlib/__init__.py", line 90, in import_module
        return _bootstrap._gcd_import(name[level:], package, level)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      File "<frozen importlib._bootstrap>", line 1387, in _gcd_import
      File "<frozen importlib._bootstrap>", line 1360, in _find_and_load
      File "<frozen importlib._bootstrap>", line 1324, in _find_and_load_unlocked
    ModuleNotFoundError: No module named 'my_workflows'

Message:

    ModuleNotFoundError: No module named 'my_workflows'