Hey gang curious if anyone has experience running dynamic sh

Question

Hey gang curious if anyone has experience running dynamic shell tasks Been banging my head against a Failed to Bind variable error for a couple hours now I realize that dynamic tasks are quite strict around what you can and can t do owing to the promises run time compilation but I just can t figure out what I m doing wrong ```from typing import List from flytekit import kwtypes workflow dynamic from flytekit extras tasks shell import OutputLocation ShellTask from flytekit types file import FlyteFile s1 = ShellTask name= shorten debug=True script= set ex head inputs infile gt outputs i inputs=kwtypes infile=FlyteFile output locs= OutputLocation var= i var type=FlyteFile location= outfile txt dynamic def shorten files files in List FlyteFile gt str for i in range len files in s1 infile=files in i return DONE workflow def wf files in List FlyteFile gt str return shorten files files in=files in if name == main print f Running wf wf ``` invoked with `pyflyte run remote dynamic shell py wf files in xaa xab ` Any help would be greatly appreciated

Pryce · Answer

Seems to be an issue with passing around FlyteFiles as doing so with a list of strings and outputting to a FlyteFile is working The salient error from before was ```Traceback most recent call last File usr local lib python3 10 site packages flytekit exceptions scopes py line 203 in user entry point return wrapped args kwargs File root cat sam py line 21 in shorten files s1 infile=files in i File usr local lib python3 10 site packages flytekit core base task py line 299 in call return flyte entity call handler self args kwargs type ignore File usr local lib python3 10 site packages flytekit core promise py line 1065 in flyte entity call handler return create and link node ctx entity=entity kwargs File usr local lib python3 10 site packages flytekit core promise py line 985 in create and link node raise AssertionError f Failed to Bind variable k for function entity name from e Message Failed to Bind variable infile for function shorten User error ``` So it seems there s some issue with passing FlyteFiles to tasks dynamically perhaps

Ketan (kumare3) · Answer

That is odd she ll tasks are no different

Ketan (kumare3) · Answer

AFK but someone will take a look

David Muraco · Answer

in the kubernetes log do you get something like this `Failed to put data from var folders dn g28 vt8518dcgmj85qjt2yr00000gq T flyte kvefddf2 raw 278a7fe38181b54b9fbfcb9a0b788243 xaa to <s3 my s3 bucket data nn f06f7de19b1384f11983 n0 0 77f44f27f51a4686d1369a43712e9d79 xaa> recursive=False n nOriginal exception Errno 2 No such file or directory var folders dn g28 vt8518dcgmj85qjt2yr00000gq T flyte kvefddf2 raw 278a7fe38181b54b9fbfcb9a0b788243 xaa n n` is it looking for a file in the job execution path instead of the relative file name

Pryce · Answer

Yep that s buried in the main flytekit ERROR log

Pryce · Answer

``` asctime 2023 04 07 05 29 07 844 name flytekit levelname ERROR message Exception when executing Failed to Bind variable infile for function shorten exc info Traceback most recent call last File usr local lib python3 10 site packages flytekit core data persistence py line 472 in put data DataPersistencePlugins find plugin remote path data config=self data config put File usr local lib python3 10 site packages flytekitplugins fsspec persist py line 129 in put return fs put from path to path recursive=recursive File usr local lib python3 10 site packages fsspec asyn py line 114 in wrapper return sync self loop func args kwargs File usr local lib python3 10 site packages fsspec asyn py line 99 in sync raise return result File usr local lib python3 10 site packages fsspec asyn py line 54 in runner result 0 = await coro File usr local lib python3 10 site packages fsspec asyn py line 503 in put return await run coros in chunks File usr local lib python3 10 site packages fsspec asyn py line 249 in run coros in chunks await asyncio gather chunk return exceptions=return exceptions File usr local lib python3 10 asyncio tasks py line 408 in wait for return await fut File usr local lib python3 10 site packages s3fs core py line 1091 in put file size = os path getsize lpath File usr local lib python3 10 genericpath py line 50 in getsize return os stat filename st size FileNotFoundError Errno 2 No such file or directory tmp flyte dxy5yx9l raw 161c980be086f6693f212f3f7f2e2e6f xaa The above exception was the direct cause of the following exception Traceback most recent call last File usr local lib python3 10 site packages flytekit core promise py line 975 in create and link node binding from python std File usr local lib python3 10 site packages flytekit core promise py line 663 in binding from python std binding data = binding data from python std ctx expected literal type t value t value type File usr local lib python3 10 site packages flytekit core promise py line 652 in binding data from python std scalar = TypeEngine to literal ctx t value t value type or type t value expected literal type scalar File usr local lib python3 10 site packages flytekit core type engine py line 820 in to literal lv = transformer to literal ctx python val python type expected File usr local lib python3 10 site packages flytekit types file file py line 330 in to literal ctx file access put data source path remote path is multipart=False File usr local lib python3 10 site packages flytekit core data persistence py line 476 in put data raise FlyteAssertion flytekit exceptions user FlyteAssertion Failed to put data from tmp flyte dxy5yx9l raw 161c980be086f6693f212f3f7f2e2e6f xaa to <s3 my s3 bucket data 3v f8099287f537c486da13 n0 0 26d802008024641c702ba75569262a56 xaa> recursive=False Original exception Errno 2 No such file or directory tmp flyte dxy5yx9l raw 161c980be086f6693f212f3f7f2e2e6f xaa The above exception was the direct cause of the following exception Traceback most recent call last File usr local lib python3 10 site packages flytekit exceptions scopes py line 203 in user entry point return wrapped args kwargs File root cat sam py line 21 in shorten files s1 infile=files in i File usr local lib python3 10 site packages flytekit core base task py line 299 in call return flyte entity call handler self args kwargs type ignore File usr local lib python3 10 site packages flytekit core promise py line 1065 in flyte entity call handler return create and link node ctx entity=entity kwargs File usr local lib python3 10 site packages flytekit core promise py line 985 in create and link node raise AssertionError f Failed to Bind variable k for function entity name from e AssertionError Failed to Bind variable infile for function shorten During handling of the above exception another exception occurred Traceback most recent call last File usr local lib python3 10 site packages flytekit core base task py line 524 in dispatch execute native outputs = self execute native inputs File usr local lib python3 10 site packages flytekit core python function task py line 166 in execute return self dynamic execute self task function kwargs File usr local lib python3 10 site packages flytekit core python function task py line 301 in dynamic execute return self compile into workflow ctx task function kwargs File usr local lib python3 10 site packages flytekit core python function task py line 196 in compile into workflow cast PythonFunctionWorkflow self wf compile kwargs File usr local lib python3 10 site packages flytekit core workflow py line 650 in compile workflow outputs = exception scopes user entry point self workflow function input kwargs File usr local lib python3 10 site packages flytekit exceptions scopes py line 214 in user entry point raise FlyteScopedUserException exc info flytekit exceptions scopes FlyteScopedUserException Failed to Bind variable infile for function shorten ```

Pryce · Answer

Do I need to run something like `create entities ` from the <https docs flyte org projects cookbook en latest auto core flyte basics shell task html sphx glr auto core flyte basics shell task py|shell task example> when the task is dynamic maybe

Pryce · Answer

I also considered that maybe passing in a list of files to `pyflyte run` isn t supported maybe the members of that list are just being parsed as strings I m still working through all this I was just curious if I others had encountered something similar

Kevin Su · Answer

I just found the issue flytekit doesn t copy the file to s3 bucket so dynamic task can t find the file I m fixing it

Pryce · Answer

Awesome thank you I ve had no trouble passing single files from the CLI args through a dynamic task FWIW

Kevin Su · Answer

Yes single file can work but list of files don t It s a bug

Pryce · Answer

Can I provide anything else for the bug ticket

Kevin Su · Answer

I created a pr <https github com flyteorg flytekit pull 1587>

Pryce · Answer

That was quick Thanks for following up I ll keep an eye on it

Pryce · Answer

As a workaround I tried using FlyteDirectory with an arbitrary path that already exists in the local object store but I keep getting No such files or directory when trying to access it via os listdir in tasks or the workflow definition itself I ve been mainly relying on the working with folders and the shell task but they instantiate FlyteDirectory with `current context working directory` so it s not exactly the same Concretely this works no problem in a task wf `x = FlyteFile ` Whereas `os listdir FlyteDirectory ` will result in No such files or directory I m guessing this is somewhat expected behavior just curious if this is a use case that should be supported or I m just doing things wrong sweat smile

Samhita Alla · Answer

It should work < Pryce> I created a folder named arbitrary in minio and when I trigger this workflow the execution s successful ```from flytekit import task workflow from flytekit types directory import FlyteDirectory task def t1 return FlyteDirectory workflow def wf t1 ```

Pryce · Answer

Hey < Samhita Alla> I gave that a go and indeed it succeeds even without the prefix existing beforehand However when I run os listdir as suggested in the Working with Folders docs then I get the No such file or directory error In this example I had made the arbitrary prefix and put a couple objects in there ```import os from typing import List from flytekit import task workflow from flytekit types directory import FlyteDirectory from flytekit types file import FlyteFile task def t1 gt List FlyteFile files = for f in os listdir FlyteDirectory files append f return files workflow def wf t1 ``` My naive suspicion is that os listdir gets called locally to the exec container and the dag doesn t convert paths for dirs like it does for full prefixes Which is why it works for `context working dir` derived paths because they re already of the form `my s3 bucket data 7z f28031d37446d44c38ec n1 0 dn3 0` but not arbitrary ones Might be way off base though Thanks for looking into it I have a workaround so I m not hung up on this particular quirk Love the blog posts btw

Samhita Alla · Answer

gt Love the blog posts btw Thanks A `FlyteDirectory` needs to be sent as an input to a task it gets uploaded downloaded at the task boundary Can you send directory as an input to `t1` and annotate it with `FlyteDirectory` `os listdir` should work then You probably might have to pass the credentials for it to work You can set the following env variables for demo cluster ```os environ AWS ENDPOINT = <http localhost 30002 > os environ AWS ACCESS KEY ID = minio os environ AWS SECRET ACCESS KEY = miniostorage ``` When I tried it was throwing an access denied error When you set the permissions correctly it has to work

Pryce · Answer

Aha I got it working the trick was passing it in at the task boundary can t believe I didn t stumble on that specific configuration in my testing I did not need credentials in the end Here is the working code if you re curious

Pryce · Answer

```import os from typing import List from flytekit import task workflow from flytekit types directory import FlyteDirectory from flytekit types file import FlyteFile task def t1 indir FlyteDirectory gt List FlyteFile files = for f in os listdir indir files append os path join indir f return files workflow def wf t1 indir=FlyteDirectory <s3 my s3 bucket arbitrary> ```

Pryce · Answer

The trick really is to pay close attention to the context that those variables are being instantiated in If it s within a task then that context won t have any knowledge of directories whereas it seems to be okay with files so dirs have to be done at the boundary Thanks for walking me through it this has been a big help

Samhita Alla · Answer

It shouldn t be okay with files either Not sure why that s working for you

Pryce · Answer

I had to look back across my commits to understand this but you re definitely right I can instantiate a FlyteDir or FlyteFile with an arbitrary path at any point within a task without issue but I can t do anything with it in that task s context So doing `x = FlyteFile path= ` won t raise an exception even though `open x r ` within that same task will raise a `file not found` error Likewise instantiating `y = FlyteDirectory path= ` will also work fine until I run `os listdir` I was never doing anything with those FlyteFiles within the same context but I assumed that since their init didn t raise an exception that their path must be valid within that context and therefore the listing out a FlyteDir should work fine

Pryce · Answer

I m just un learning my local filesystem intuitions assumptions that are evidently pretty baked in sweat smile