Hi Trying to understand how I can use prefect with dask whil

Question

Hi Trying to understand how I can use prefect with dask while still debug locally I am getting an error More details code and errors inside thread

Tomer Cagan · Answer

A few notes I am using `worker client` inside a task to be able to do my own dynamic parallelization a requirement for the algorithm that I run I saw in the dask documentation that it is possible to specify ` single threaded ` scheduler for debugging and this is my end goal with doing it I have the following code ```import time import random from typing import List from prefect import Flow task Parameter from prefect storage import Docker from prefect executors import DaskExecutor LocalDaskExecutor from dask distributed import worker client def fib value int gt int delay = random random 2 print f calculating fib value sleeping delay seconds to simulate slow processing time sleep delay if value lt 2 return value with worker client as client a future = client submit fib value 1 b future = client submit fib value 2 a b = client gather a future b future return a + b task log stdout=True def compute with dask values List int gt int max value = max values print f got a list of size len values starting long process to calculate fib max value results = fib max value print f got results of calculation the value is results task def load stuff return 1 2 3 4 with Flow fib dask flow as flow loaded = load stuff compute with dask loaded flow storage = Docker registry url= <http docker k8s nextsilicon com repository container ns |docker k8s nextsilicon com repository container ns > image name= research dask fib flow executor = DaskExecutor if name == main flow executor = LocalDaskExecutor synchronous flow storage = None flow run print running main ``` And I am getting the following error ``` python flow recur py 2022 03 10 10 42 33+0200 INFO prefect TaskRunner | Task load stuff Starting task run 2022 03 10 10 42 33+0200 INFO prefect TaskRunner | Task load stuff Finished task run for task with final state Success 2022 03 10 10 42 33+0200 INFO prefect TaskRunner | Task compute with dask Starting task run 2022 03 10 10 42 33+0200 INFO prefect TaskRunner | got a list of size 4 starting long process to calculate fib 11 2022 03 10 10 42 33+0200 INFO prefect TaskRunner | calculating fib 4 sleeping 1 8652929878459892 seconds to simulate slow processing 2022 03 10 10 42 35+0200 ERROR prefect TaskRunner | Task compute with dask Exception encountered during task execution Traceback most recent call last File Users tomercagan dev prefect play venv lib python3 9 site packages distributed worker py line 4219 in get worker return thread state execution state worker AttributeError thread local object has no attribute execution state During handling of the above exception another exception occurred Traceback most recent call last File Users tomercagan dev prefect play venv lib python3 9 site packages distributed worker py line 4222 in get worker return first File Users tomercagan dev prefect play venv lib python3 9 site packages toolz itertoolz py line 376 in first return next iter seq StopIteration During handling of the above exception another exception occurred Traceback most recent call last File Users tomercagan dev prefect play venv lib python3 9 site packages prefect engine task runner py line 876 in get task run state value = prefect utilities executors run task with timeout File Users tomercagan dev prefect play venv lib python3 9 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File Users tomercagan dev prefect play flow recur py line 40 in compute with dask results = fib max value File Users tomercagan dev prefect play flow recur py line 17 in fib with worker client as client File usr local Cellar python 3 9 3 9 10 Frameworks Python framework Versions 3 9 lib python3 9 contextlib py line 119 in enter return next self gen File Users tomercagan dev prefect play venv lib python3 9 site packages distributed worker client py line 52 in worker client worker = get worker File Users tomercagan dev prefect play venv lib python3 9 site packages distributed worker py line 4228 in get worker raise ValueError No workers found ValueError No workers found 2022 03 10 10 42 35+0200 INFO prefect TaskRunner | Task compute with dask Finished task run for task with final state Failed 2022 03 10 10 42 35+0200 INFO prefect FlowRunner | Flow run FAILED some reference tasks failed running main prefect play ``` This happens to me whether I am using ` syncronous ` or leave the `LocalDaskCluster` constructor empty I am guessing that I am doing something wrong because there s not much there Also running this as a registered flow with Prefect Server it works fine

Tomer Cagan · Answer

This does work ```if name == main print running main executor = DaskExecutor flow storage = None flow run executor=executor ``` I am also able to stop in breakpoint visual studio code but it is hard to debug due to multiple threads running

Anna Geller · Answer

I may be wrong here but I believe that attaching Dask executor to your flow matters for mapping running tasks independent of each other in parallel If you implement your own worker client within your Prefect task you 1 Don t use Prefect s paralellism which is why I think the choice of executor likely doesn t matter much in this example flow you should see the same behavior even without attaching any executor 2 Lose observability that Prefect provides to each mapped child task when parallelizing your processing which may be a totally valid option in your use case just bringing it up for you to know If you run this script with the `fib` function without Prefect I think you should see the same error LMK if this is not true But if it s true and it s purely a Dask issue you are dealing with you may get a smarter response that I can give you if you post your question in the slightly smiling face Lastly when it comes to things like your Fibonacci example actually you may be better off doing something like rather than parallel processing

Tomer Cagan · Answer

If I change my main to this ```if name == main client = Client n workers= 1 threads per worker = 1 future = client submit fib 10 print future result ``` It works fine and I can debug so I don t think it is a dask issue I believe it is something wrong with how I initialize the dask cluster Also with `DaskExecutor` it works fine and I can also debug

Tomer Cagan · Answer

I understand the issues you brought up As I stated before the algorithm we are trying to orchestrate is kind of dynamic and recursive and I am not sure that task looping would work in this case

Tomer Cagan · Answer

BTW in the code above I am not implementing my own worker client I am piggy backing on the dask cluster already created by Prefect So it will be running additional tasks in parallel to the Prefect tasks at least as far as I understand

Anna Geller · Answer

Gotcha I remember we had this conversation before and I remember recommending resource manager for such use cases Usually users who need access to the Dask client directly initialize it with resource manager but I agree that given you initialize this within a task it shouldn t matter Perhaps running this within a Docker container is the issue Sometimes it s hard to allocate enough resources from the VM to the container and you may need to change your Docker settings or use host config ```DockerRun host config= xxx xxx ``` Maybe indeed for such use cases you re better off using Orion directly since you could simply run this directly in your flow ``` flow def your flow function client = Client n workers= 1 threads per worker = 1 future = client submit fib 10 print future result ```

Tomer Cagan · Answer

Yeah we ve kind of carried out this conversation over several threads I am not running on docker there it is running locally note how in the main I reset the storage and I just do `flow run ` I took your advice and did some of my own tests and this flow that I currently have works nicely on Prefect 1 0 and sort of giving me what I wanted minus the extra observability that Prefect provides In the example you provided I think that flow will spawn its own cluster which is not what I am trying to do As for Orion I like a lot of the new ideas there but I am not sure whether I will be able to sell it to the team especially since right now in order to piggyback onto the prefect spawned dask cluster I would have to use async which the team here does not like so much plus it being in beta stages I ought to write some article at the end of my discovery process slightly smiling face

Anna Geller · Answer

Writing an article about your experience would be fantastic pray Regarding Docker storage even if you don t override it it won t be used when doing `flow run ` since storage is built on registration not at runtime Docker was a red herring here slightly smiling face Just to be sure we are on the same page Orion client is async but you don t have to use async tasks flows and you don t need to know anything about async to use Orion especially if you use your custom Python code to achieve parallelism in the flow and you don t rely on any prebuilt task runners And for Prefect 1 0 given your use case I wonder whether spinning up a long running standalone Dask cluster may help in your use case this way regardless of whether you run it locally or triggering a backend run you may leverage the same client configuration in your task There are some tradeoffs here but just making sure you are aware of that option as well <https discourse prefect io t are there any guidelines for using a temporary vs static dask cluster how to set one on kubernetes 342 2 h 3 connecting to an existing cluster 5|https discourse prefect io t are there any guidelines for using a temporary vs static dask cluster how to set one on kubernetes 342 2 h 3 ter 5>

Tomer Cagan · Answer

Hi About Orion I tried to run the same code there and got an error that and according to < Zanie> right now I would have to use an async task in order for it to work and everything this entails About long running cluster I don t know if this would really help me here it won t lend itself for local debugging and one of the issues we have with our current system is that tasks interfere with each other causing starvation and sometime crushes of the whole system We are starting to migrate to k8s and I like the idea of ephemeral dask clusters I think at the end of the day I can probably configure a `DaskExecutor` to run locally and with a single worker and debug using it

Anna Geller · Answer

You re definitely right if your workloads are competing for resources then using temporary cluster e g on Kubernetes is a smart choice Sorry didn t know about the async issue you mentioned I need to catch up sweat smile

Kevin Kho · Answer

I was reading and I think the issue might be the submission of `a future` and `b future` simultaneously trying to read value from different workers Very unsure But I think this might be more stable with processes as the scheduler It looks like the threads are hitting each other so you d otherwise need the Really unsure

Tomer Cagan · Answer

Mmmm I believe that with `client submit` is creates a `Future` and not `Delayed` thought tbh I am not exactly sure what the difference is wink I will try to look understand the SO thread more deeply and maybe I can learn from but I am not using locks explicitly Also I am confused why is works ok with `DaskExecutor` and not with `LocalDaskExecutor` with and without synchronous Is there a way to configure `DaskExecutor` for a single worker single thread I guess that could possible help me in debugging in the general sense not this issue specifically

Anna Geller · Answer

gt I am confused why is works ok with `DaskExecutor` and not with `LocalDaskExecutor` with and without synchronous `LocalDaskExecutor` relies on multithreading and multiprocessing without creating a local Dask cluster In contrast `DaskExecutor` creates a local Dask cluster <https discourse prefect io t what is the difference between a daskexecutor and a localdaskexecutor 374>

Tomer Cagan · Answer

Wow didn t expect an answer until tomorrow afternoon Thanks for the info I will review and see whether it somehow lead me in a good direction

Tomer Cagan · Answer

This sort of works for debugging ```if name == main print running main executor = DaskExecutor debug=True cluster kwargs= n workers 1 flow storage = None flow run executor=executor ``` Sometimes it seems to get stuck in case I am stopping running many time with a breakpoint

Anna Geller · Answer

nice btw no need to set storage to None storage and run config are ignored when you do `flow run `

Tomer Cagan · Answer

That s good information thanks Part of what I am trying to understand is how to manage the development to easily allow the researchers to be able to seamlessly move from local development and debugging to running automated scheduled tests of the code

Madhup Sukoon · Answer

Hi I was setting up a flow and got this error Anyone know what exit code 126 is

Zanie · Answer

gt If a command is found but is not executable the return status is 126

Madhup Sukoon · Answer

Could you elaborate a little please or point me to the documentation

Madhup Sukoon · Answer

Also how do i know which command is causing this

Zanie · Answer

I just googled the exit code smile

Zanie · Answer

These are standard linux exit codes not Prefect related

Zanie · Answer

What image are you using

Madhup Sukoon · Answer

Oh I see thanks I am using a custom docker image that I just created

Zanie · Answer

To run a flow we call a `prefect flow run execute` in the container

Madhup Sukoon · Answer

Okay let me try running that manually in the container

Zanie · Answer

You can try it locally with `docker run t lt your image gt prefect flow run execute`

Zanie · Answer

thumbsup

Madhup Sukoon · Answer

Thanks a lot