Hello there! 👋 Question on handling files/directories on s3/gcs for performance. In all the examples I've seen about reading files from a remote directory, we define workflows as multiple tasks. To my understanding, generally the first one yields a FlyteDirectory, and later tasks work on the content that was downloaded locally (tmp/...). For instance, when reading from a bucket containing two txt files (

foo.txt

and

bar.txt

):

import os
from flytekit import task, workflow, Resources
from flytekit.types.file import FlyteFile
from flytekit.types.directory import FlyteDirectory
from typing import Tuple

@workflow
def wf(dirpath: str) -> str:

    # These two below works, as get_dir downloads the directory content to the local disk.
    flytedir = get_dir(dirpath=dirpath)
    foo, bar = do_stuff_on_dir(flytedir=flytedir)

    hello = say_hello(foo=foo, bar=bar)

    return hello

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

@task()
def do_stuff_on_dir(flytedir: FlyteDirectory) -> Tuple[str, str]:
    texts = []

    for fname in os.listdir(flytedir):
        filepath = os.path.join(flytedir, fname)

        with open(filepath, 'r') as f:
            content = f.read()
        texts.append(content)

    # Do stuff
    texts = [texts[1], texts[0]]

    return texts

@task(requests=Resources(cpu="1", mem="1000Mi", ephemeral_storage="2000Mi"), limits=Resources(cpu="2", mem="1000Mi", ephemeral_storage="3000Mi"))
def say_hello(foo: str, bar: str) -> str:
    return f"Hello, {foo} and {bar}!"

However when run on a cluster (deployment on a single cluster) this comes with a k8s overhead for spawning the temporary pods for each task. To speed up the workflow I was thinking about reducing the overhead by aggregating tasks, reading files from the bucket and doing stuff on them in a single task. Is there a preferred way to do it, effectively getting to the same local tmp directory structure before returning a FlyteDirectory as a task output? For instance something like the following chunk works in sandbox but not on my single cluster deployment (exploiting

FlyteDirectory.listdir(flytedir)

)

import os
from flytekit import task, workflow, Resources
from flytekit.types.file import FlyteFile
from flytekit.types.directory import FlyteDirectory
from typing import Tuple


@workflow
def wf(dirpath: str) -> str:

    foo, bar = get_dir_and_do_stuff(dirpath=dirpath)

    hello = say_hello(foo=foo, bar=bar)

    return hello

@task()
def get_dir_and_do_stuff(dirpath: str) -> Tuple[str, str]:
    
    flytedir = FlyteDirectory(path=dirpath)

    texts = []
    # for fname in os.listdir(flytedir): # This works only locally (not even sandbox)
    for fname in flytedir.listdir(flytedir):  # This works on sandbox but not on cluster

        filepath = os.path.join(flytedir, fname)
        infile = FlyteFile(path=filepath)
        with open(infile, 'r') as f:
            content = f.read()
        texts.append(content)
        
    # Do stuff
    # ...

    return texts

@task()
def say_hello(foo: str, bar: str) -> str:
    return f"Hello, {foo} and {bar}!"

I'd like to avoid the alternative of having a single task which instead of exploiting FlyteDirectories just uses the GS client to read from a bucket. Any other tip for optimizing workflows for speed? Thanks!

This is an awesome question! There are a few things to unpack here. #1 As I'm sure you are aware, flyte tasks are always stateless (part of the system guarantee for safety to ensure it can cache/retry... etc.). That means when a task returns a FlyteDirectory, what actually happens is that it uploads the contents of that local directory to GCS. And depending on how big the directory is that might be an expensive operation... One thing I would optimize away is removing

get_dir

altogether.

@workflow
def wf(flytedir: FlyteDirectory) -> str:

    foo, bar = do_stuff_on_dir(flytedir=flytedir)

    hello = say_hello(foo=foo, bar=bar)

    return hello

This can continue to be invoked locally with

--flytedir /my-local/path

And it will correctly behave whether you run locally (will just operate on the local paths) or try to launch it on a cluster (will automatically upload the local path to remote GCS then pass the GCS path around)

Thanks Haytham for your detailed answer! About #1: that makes sense if files are in a local path from the start. But what for reading stuff which is exclusively on a bucket? Defining the workflow input as a FlyteDirectory and providing a bucket uri instead would result in something like

Invalid value for '--flytedir': parameter should be a valid directory path, <s3://my-s3-bucket/foobar>

, or am I missing something? Can we skip get_dir also without a local path to provide?

i can repro the `parameter should be a valid directory path`… agree that’s a bit odd, we’ll take a look

Thanks @thankful-minister-83577!

https://github.com/flyteorg/flytekit/pull/2547 was fixed. it’ll get pushed out with the next release.

Thank you @thankful-minister-83577! 🙏🏻 I will test this again once released then, is there a planned release calendar that you know?

at the very least, we can get a beta release out for flytekit today.

@brash-autumn-99506 ICYMI this is the release: https://github.com/flyteorg/flytekit/releases/tag/v1.13.0

Thank you @average-finland-92144! Tested, it works 🙌🚀