I am trying to get my flow to run in a remote Ray anyscale c

Question

I am trying to get my flow to run in a remote Ray anyscale cluster However my flow is failing whenever I try to submit a task The error is that a Ray worker is trying to access something that is specific to my local machine I am running directly on my local machine for development I think this has something to do with local storage being used even in a distributed context In the doc for RayTaskRunner it recommends to use remote storage I setup an S3 block but then I have no idea what I suppose to do next My flow is still using Local storage ```from prefect import flow task get run logger from prefect ray import RayTaskRunner task def generate numbers n int gt list int return list range n task def double the numbers d int gt int return d 2 flow task runner=RayTaskRunner address= init kwargs= runtime env working dir def run workflow logger = get run logger Hello World numbers = generate numbers submit 100 numbers result doubled = double the numbers map numbers d wait result for d in doubled run workflow ``` ```ray begin task run pid=20006 ip= File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect utilities asyncutils py line 193 in wrapper File home ray anaconda3 lib python3 10 site packages prefect utilities asyncutils py line 140 in run async in new loop return anyio run partial fn args kwargs File home ray anaconda3 lib python3 10 site packages anyio core eventloop py line 70 in run return asynclib run func args backend options File home ray anaconda3 lib python3 10 site packages anyio backends asyncio py line 292 in run return native run wrapper debug=debug File home ray anaconda3 lib python3 10 asyncio runners py line 44 in run return loop run until complete main File home ray anaconda3 lib python3 10 asyncio base events py line 646 in run until complete return future result File home ray anaconda3 lib python3 10 site packages anyio backends asyncio py line 287 in wrapper return await func args File home ray anaconda3 lib python3 10 site packages prefect engine py line 946 in begin task run return await orchestrate task run File home ray anaconda3 lib python3 10 site packages prefect engine py line 1083 in orchestrate task run await persist serialized result File home ray anaconda3 lib python3 10 site packages prefect results py line 15 in persist serialized result await filesystem write path key content File home ray anaconda3 lib python3 10 site packages prefect filesystems py line 162 in write path path parent mkdir exist ok=True parents=True File home ray anaconda3 lib python3 10 pathlib py line 1177 in mkdir self parent mkdir parents=True exist ok=True File home ray anaconda3 lib python3 10 pathlib py line 1177 in mkdir self parent mkdir parents=True exist ok=True File home ray anaconda3 lib python3 10 pathlib py line 1177 in mkdir self parent mkdir parents=True exist ok=True File home ray anaconda3 lib python3 10 pathlib py line 1173 in mkdir self accessor mkdir self mode PermissionError Errno 13 Permission denied Users ``` < Taylor Curran> < Kalise Richmond>

Christopher Boyd · Answer

Hi Kha Looks like you are missing the imports for storage blocks and usage <https docs prefect io api ref prefect filesystems prefect filesystems S3>

Christopher Boyd · Answer

It would look like this in your flow ```from prefect filesystems import S3 s3 block = S3 load BLOCK NAME ```

Christopher Boyd · Answer

Also would you mind condensing your code blocks into this thread in lieu of multiple entries in the channel

Kha Nguyen · Answer

Oh my bad I forgot to create a thread

Kha Nguyen · Answer

Does it mean that when I run S3 load the filesystem will use S3 from then on

Kha Nguyen · Answer

I put that code in and nothing changes

Kha Nguyen · Answer

```from prefect import flow task get run logger from prefect ray import RayTaskRunner from prefect filesystems import S3 task def generate numbers n int gt list int return list range n task def double the numbers d int gt int return d 2 flow task runner=RayTaskRunner address= init kwargs= runtime env working dir def run workflow s3 block = S3 load demo logger = get run logger Hello World numbers = generate numbers submit 100 numbers result doubled = double the numbers map numbers d wait result for d in doubled run workflow ```

Christopher Boyd · Answer

So there are a couple things 1 I don t think your s3 block needs to be in the flow definition 2 Are you building a deployment then executing this from a deployment 3 When you build and apply that deployment are you passing the storage block <https orion docs prefect io concepts storage >

Kha Nguyen · Answer

No this is run directly on my local computer I will build a deployment when the flow is ready

Christopher Boyd · Answer

If you are running this locally on your computer without a deployment the storage block is not being used

Kha Nguyen · Answer

So the only way to use block storage is to run from a deployment right

Kha Nguyen · Answer

Is there a way to tell my local process to use the block storage because I think my local machine is simply acting as an agent

Christopher Boyd · Answer

Christopher Boyd · Answer

For the RayTask runner I think the assumption here is that you have specified your storage block and consequently that deployment where it will be packaged

Christopher Boyd · Answer

What is this address ```address= <anyscale singularity demo1> ```

Christopher Boyd · Answer

I think if your intent is to run this locally then the address should be removed to spin up a local instance If the intent is to point to another instance then a storage block should be configure with the deployment

Kha Nguyen · Answer

Anyscale for Ray is similar to Coiled for Dask Whenever I init a Ray cluster pointing to this anyscale address then Anyscale will provision a cluster for me

Kha Nguyen · Answer

My intention is to run locally while using a remote Ray cluster

Kha Nguyen · Answer

Because during development I do not wish to make deployment after minor adjustments I will want to use the power of distributed Ray however

Kha Nguyen · Answer

I just made a deployment for the sake of using the remote file system S3 block and I am still seeing the same error

Kha Nguyen · Answer

```23 10 48 652 | INFO | prefect agent Submitting flow run b6b7609b 3936 4b37 a70a 2ef069c9a3ad 23 10 49 243 | INFO | prefect infrastructure process Opening process literate honeybee 23 10 49 262 | INFO | prefect agent Completed submission of flow run b6b7609b 3936 4b37 a70a 2ef069c9a3ad 23 10 51 796 | INFO | Flow run literate honeybee Starting RayTaskRunner submitted tasks will be run in parallel 23 10 51 797 | INFO | prefect task runner ray Connecting to an existing Ray instance at Authenticating Loaded Anyscale authentication token from ~ anyscale credentials json Parsing Ray Client arguments Warning No working dir specified Files will only be uploaded to the cluster if a working dir is provided or a project is detected In the future files will only be uploaded if working dir is provided To ensure files continue being imported going forward set the working dir in your runtime environment See Finished parsing arguments Choosing a project Using the project singularity name singularity project id prj 9dVsTxDmpWhannc9eLbHDaPv Preparing the cluster Cluster demo1 is currently running Connecting to this cluster cluster id ses PXJLgBvwGrAM4DaFij4MaGdM cluster environment apt ufz5Tj2dgkhiZXVmHvqR48DH 3 cluster environment id bld ne2CdBTrfiUBsMtLqpW1XxUw cluster compute singularity cluster compute id cpt T4DnCqrTzVrqnhq4FUdmWfyw idle termination 120 minutes maximum uptime disabled link Starting the interactive session Connected to demo1 Interactive session engine py 08 03 2022 20 10 58 has started interactive session id job 3etBZ5DVnCp3jmmsYR82U4k7 runtime environment link 23 11 19 847 | INFO | prefect task runner ray Using Ray cluster with 1 nodes 23 11 19 847 | INFO | prefect task runner ray The Ray UI is available at 23 11 21 020 | INFO | Flow run literate honeybee Hello World 23 11 21 344 | INFO | Flow run literate honeybee Created task run generate numbers 92c16399 0 for task generate numbers 23 11 22 566 | INFO | Flow run literate honeybee Submitted task run generate numbers 92c16399 0 for execution 23 11 24 210 | ERROR | Flow run literate honeybee Encountered exception during execution Traceback most recent call last File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect engine py line 557 in orchestrate flow run result = await run sync flow call File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect utilities asyncutils py line 56 in run sync in worker thread return await anyio to thread run sync call cancellable=True File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages anyio to thread py line 31 in run sync return await get asynclib run sync in worker thread File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages anyio backends asyncio py line 937 in run sync in worker thread return await future File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages anyio backends asyncio py line 867 in run result = context run func args File demo1 py line 27 in run workflow numbers result File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect futures py line 209 in result return sync File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect utilities asyncutils py line 221 in sync return run async from worker thread async fn args kwargs File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect utilities asyncutils py line 136 in run async from worker thread return anyio from thread run call File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages anyio from thread py line 49 in run return asynclib run async from thread func args File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages anyio backends asyncio py line 970 in run async from thread return f result File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 concurrent futures base py line 446 in result return self get result File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 concurrent futures base py line 391 in get result raise self exception File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect futures py line 220 in result return final state result raise on failure=raise on failure File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect orion schemas states py line 145 in result raise data File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect ray task runners py line 172 in wait result = await ref ray exceptions RayTaskError ray begin task run pid=2225 ip=172 31 38 127 FileNotFoundError Errno 2 No such file or directory Users kha prefect storage During handling of the above exception another exception occurred ray begin task run pid=2225 ip=172 31 38 127 File home ray anaconda3 lib python3 10 pathlib py line 1173 in mkdir self accessor mkdir self mode FileNotFoundError Errno 2 No such file or directory Users kha prefect During handling of the above exception another exception occurred ray begin task run pid=2225 ip=172 31 38 127 File home ray anaconda3 lib python3 10 pathlib py line 1173 in mkdir self accessor mkdir self mode FileNotFoundError Errno 2 No such file or directory Users kha During handling of the above exception another exception occurred ray begin task run pid=2225 ip=172 31 38 127 File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect utilities asyncutils py line 193 in wrapper File home ray anaconda3 lib python3 10 site packages prefect utilities asyncutils py line 140 in run async in new loop return anyio run partial fn args kwargs File home ray anaconda3 lib python3 10 site packages anyio core eventloop py line 70 in run return asynclib run func args backend options File home ray anaconda3 lib python3 10 site packages anyio backends asyncio py line 292 in run return native run wrapper debug=debug File home ray anaconda3 lib python3 10 asyncio runners py line 44 in run return loop run until complete main File home ray anaconda3 lib python3 10 asyncio base events py line 646 in run until complete return future result File home ray anaconda3 lib python3 10 site packages anyio backends asyncio py line 287 in wrapper return await func args File home ray anaconda3 lib python3 10 site packages prefect engine py line 946 in begin task run return await orchestrate task run File home ray anaconda3 lib python3 10 site packages prefect engine py line 1083 in orchestrate task run await persist serialized result File home ray anaconda3 lib python3 10 site packages prefect results py line 15 in persist serialized result await filesystem write path key content File home ray anaconda3 lib python3 10 site packages prefect filesystems py line 162 in write path path parent mkdir exist ok=True parents=True File home ray anaconda3 lib python3 10 pathlib py line 1177 in mkdir self parent mkdir parents=True exist ok=True File home ray anaconda3 lib python3 10 pathlib py line 1177 in mkdir self parent mkdir parents=True exist ok=True File home ray anaconda3 lib python3 10 pathlib py line 1177 in mkdir self parent mkdir parents=True exist ok=True File home ray anaconda3 lib python3 10 pathlib py line 1173 in mkdir self accessor mkdir self mode PermissionError Errno 13 Permission denied Users ```

Kha Nguyen · Answer

I wonder if we are talking about the same thing The S3 block that you referred to seem to be used to store the python code to be used for execution

Kha Nguyen · Answer

My issue at the moment is why this simple task cannot run on a Ray cluster We have a task as simple as it possibly can be and as soon as a submit request is made it starts to look for something on my local computer Users kha prefect storage

Kha Nguyen · Answer

For completeness here was how a deployment was made ```prefect deployment build demo1 py run workflow n demo t demo i process sb s3 demo prefect deployment apply run workflow deployment yaml``` ``` A complete description of a Prefect Deployment for flow run workflow name demo description a demo deployment tags demo parameters schedule null infrastructure type process env labels name null command python m prefect engine stream output true DO NOT EDIT BELOW THIS LINE flow name run workflow manifest path run workflow manifest json storage bucket path xxx aws access key id aws secret access key is anonymous true block document name anonymous 768c2c8b 7b65 4281 b6b4 c549446cf49a block document id 262d5c8e e649 4aa2 afa3 ee17a8813efa block type slug s3 parameter openapi schema title Parameters type object properties required null definitions null```

Andrew Huang · Answer

Are you able to run this without providing any address keywords ```import time from prefect import flow task from prefect ray import RayTaskRunner task def shout number time sleep 0 5 print f number flow task runner=RayTaskRunner def count to highest number for number in range highest number shout submit number if name == main count to 10 ``` Or drop `init kwargs= runtime env working dir ` in your example

Kha Nguyen · Answer

It works when I run everything locally Which was my original result Then when I moved to Prefect 2 0 Cloud there was some issue with importing packaging requirements even though I installed that the package packaging Then when I moved to a remote Ray cluster which is how we run in production so just because a local toy example works it is not production ready when a task is submitted I have traced the issue to the line of code when a result is trying to persist it still uses LocalFileSystem I think this problem should be solved when we use a remote file system to persist task outputs However in Prefect documentation the storage or filesystem is used to described both storage for flow code and storage for task output which is confusing And then the document went on to say that we should use remote filesystem for distributed computing context and then show an example of creating an S3 block to store flow code man shrugging

Kha Nguyen · Answer

What I really need at this time is to tell the flow when running to use S3 filesystem instead of LocalFileSystem I hope someone < Taylor Curran> < Kalise Richmond> < Christopher Boyd> can answer if this is possible a yes no question at the moment

Kha Nguyen · Answer

In case you are wondering this line of code was hit when running using remote Ray cluster ```File home ray anaconda3 lib python3 10 site packages prefect filesystems py line 162 in write path path parent mkdir exist ok=True parents=True ```

Christopher Boyd · Answer

Hi Kha Just so i m following Your initial issue was ``` self accessor mkdir self mode PermissionError Errno 13 Permission denied Users ``` This was resulting from using a remote Ray anyscale cluster without a remote storage block and it did not have access to your local filesystem Now your latest is that is does run successfully locally but does not run remote in anyscale when using a storage block Is that accurate

Kha Nguyen · Answer

Hi Christopher yes that is accurate

Christopher Boyd · Answer

What is the full stack trace of the latest error you are receiving and the command you used to register the deployment storage block

Kha Nguyen · Answer

The S3 storage block was created in the UI named `demo` The deployment was made as follow ```prefect deployment build demo1 py run workflow n demo t local i process sb s3 demo prefect deployment apply run workflow deployment yaml```

Kha Nguyen · Answer

Full stack trace as soon as a task is submitted ```Traceback most recent call last File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect engine py line 557 in orchestrate flow run result = await run sync flow call File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect utilities asyncutils py line 56 in run sync in worker thread return await anyio to thread run sync call cancellable=True File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages anyio to thread py line 31 in run sync return await get asynclib run sync in worker thread File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages anyio backends asyncio py line 937 in run sync in worker thread return await future File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages anyio backends asyncio py line 867 in run result = context run func args File demo1 py line 27 in run workflow <http logger info|logger info> numbers result File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect futures py line 209 in result return sync File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect utilities asyncutils py line 221 in sync return run async from worker thread async fn args kwargs File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect utilities asyncutils py line 136 in run async from worker thread return anyio from thread run call File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages anyio from thread py line 49 in run return asynclib run async from thread func args File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages anyio backends asyncio py line 970 in run async from thread return f result File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 concurrent futures base py line 446 in result return self get result File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 concurrent futures base py line 391 in get result raise self exception File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect futures py line 220 in result return final state result raise on failure=raise on failure File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect orion schemas states py line 145 in result raise data File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect ray task runners py line 172 in wait result = await ref ray exceptions RayTaskError ray begin task run pid=2225 ip=172 31 38 127 FileNotFoundError Errno 2 No such file or directory Users kha prefect storage During handling of the above exception another exception occurred ray begin task run pid=2225 ip=172 31 38 127 File home ray anaconda3 lib python3 10 pathlib py line 1173 in mkdir self accessor mkdir self mode FileNotFoundError Errno 2 No such file or directory Users kha prefect During handling of the above exception another exception occurred ray begin task run pid=2225 ip=172 31 38 127 File home ray anaconda3 lib python3 10 pathlib py line 1173 in mkdir self accessor mkdir self mode FileNotFoundError Errno 2 No such file or directory Users kha During handling of the above exception another exception occurred ray begin task run pid=2225 ip=172 31 38 127 File opt homebrew Caskroom miniforge base envs singularity2 lib python3 10 site packages prefect utilities asyncutils py line 193 in wrapper File home ray anaconda3 lib python3 10 site packages prefect utilities asyncutils py line 140 in run async in new loop return anyio run partial fn args kwargs File home ray anaconda3 lib python3 10 site packages anyio core eventloop py line 70 in run return asynclib run func args backend options File home ray anaconda3 lib python3 10 site packages anyio backends asyncio py line 292 in run return native run wrapper debug=debug File home ray anaconda3 lib python3 10 asyncio runners py line 44 in run return loop run until complete main File home ray anaconda3 lib python3 10 asyncio base events py line 646 in run until complete return future result File home ray anaconda3 lib python3 10 site packages anyio backends asyncio py line 287 in wrapper return await func args File home ray anaconda3 lib python3 10 site packages prefect engine py line 946 in begin task run return await orchestrate task run File home ray anaconda3 lib python3 10 site packages prefect engine py line 1083 in orchestrate task run await persist serialized result File home ray anaconda3 lib python3 10 site packages prefect results py line 15 in persist serialized result await filesystem write path key content File home ray anaconda3 lib python3 10 site packages prefect filesystems py line 162 in write path path parent mkdir exist ok=True parents=True File home ray anaconda3 lib python3 10 pathlib py line 1177 in mkdir self parent mkdir parents=True exist ok=True File home ray anaconda3 lib python3 10 pathlib py line 1177 in mkdir self parent mkdir parents=True exist ok=True File home ray anaconda3 lib python3 10 pathlib py line 1177 in mkdir self parent mkdir parents=True exist ok=True File home ray anaconda3 lib python3 10 pathlib py line 1173 in mkdir self accessor mkdir self mode PermissionError Errno 13 Permission denied Users ```