Hi slightly smiling face I m facing an error on an HPC clust

Question

Hi slightly smiling face I m facing an error on an HPC cluster with my hello world only with prefect 2

Thomas Mignon · Answer

Here is my functionnal hello world with prefect 1 ```import time from dask jobqueue import PBSCluster from prefect executors import DaskExecutor from prefect import Flow task task def times return 1 5 10 task def sleep x print f sleeping for x time sleep x print f done sleeping for x return x with Flow mapping test as flow sleep map sleep map times if name == main flow run executor=DaskExecutor cluster class=PBSCluster cluster kwargs= cores 1 memory 24GB queue sequentiel walltime 00 05 00 interface ib0 n workers 1 ``` launched with `python myhelloworld py` i got an output file dask worker o with my prints the qsub was done well by dask on the PBS cluster and if i do the same helloworld with prefect 2 ```import time from prefect import flow task from prefect task runners import DaskTaskRunner task def times return 1 5 10 task def sleep x print f sleeping for x time sleep x print f done sleeping for x return x if name == main flow task runner=DaskTaskRunner cluster class= dask jobqueue PBSCluster cluster kwargs= cores 1 memory 24GB queue sequentiel walltime 00 05 00 interface ib0 n workers 1 def hello world flow sleep map sleep map times ``` I have no output file at the location where i launch my python script i have doubt that the qsub was really submitted or maybe there is an option that has been set in prefect 2 to set the output in another location I m a bit lost slightly smiling face

Anna Geller · Answer

The map operator doesn t exist yet in Prefect 2 0 You could do ```from prefect import flow task from prefect task runners import DaskTaskRunner task def my mapped task x return x task def my mapped task 2 x return x task def list task return 1 2 3 4 5 flow task runner=task runner=DaskTaskRunner cluster class= dask jobqueue PBSCluster cluster kwargs= cores 1 memory 24GB queue sequentiel walltime 00 05 00 interface ib0 n workers 1 def my flow for i in range 10 my mapped task i ```

Anna Geller · Answer

and then it s best to then call the flow within main ```if name == main my flow ```

Thomas Mignon · Answer

my code is now ```from prefect import flow task from prefect task runners import DaskTaskRunner task def my mapped task x print x if name == main flow task runner=DaskTaskRunner cluster class= dask jobqueue PBSCluster cluster kwargs= cores 1 memory 24GB queue sequentiel walltime 00 05 00 interface ib0 n workers 1 def hello world flow for i in range 10 my mapped task i ``` But this is not better i don t have any output file from dask

Anna Geller · Answer

why do you keep defining your flow in main smile can you try ```from prefect import flow task from prefect task runners import DaskTaskRunner task def my mapped task x print x flow task runner=DaskTaskRunner cluster class= dask jobqueue PBSCluster cluster kwargs= cores 1 memory 24GB queue sequentiel walltime 00 05 00 interface ib0 n workers 1 def hello world flow for i in range 10 my mapped task i if name == main hello world flow ``` main is a gate to start some process here to trigger a flow run from the flow function Your flow shouldn t be in main this way you couldn t pass it e g to DeploymentSpec what do you mean by output file from Dask

Anna Geller · Answer

by default when you run this you should see a folder dask worker space is this what you re looking for

Thomas Mignon · Answer

the hello world worked with your code slightly smiling face

Anna Geller · Answer

Nice raised hands

Thomas Mignon · Answer

But where i m supposed to see the print

Thomas Mignon · Answer

In my file dask worker o i have

Thomas Mignon · Answer

```distributed nanny INFO Start Nanny at distributed worker INFO Start worker at distributed worker INFO Listening to distributed worker INFO dashboard at 10 148 0 181 40645 distributed worker INFO Waiting to connect to distributed worker INFO distributed worker INFO Threads 1 distributed worker INFO Memory 22 35 GiB distributed worker INFO Local Directory dev shm pbs 9835088 datarmor0 dask worker space worker apnm7ik1 distributed worker INFO distributed worker INFO Registered to distributed worker INFO distributed core INFO Starting established connection distributed core INFO Event loop was unresponsive in Worker for 12 74s This is often caused by long running GIL holding functions or moving large chunks of data This can cause timeouts and instability home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect client py 1195 UserWarning No default storage has been set on the server Using temporary local storage for results warnings warn 13 25 49 297 | INFO | Task run my mapped task ca1f3dbb 0 Finished in state Completed None 13 25 51 325 | INFO | Task run my mapped task ca1f3dbb 1 Finished in state Completed None 13 25 53 381 | INFO | Task run my mapped task ca1f3dbb 2 Finished in state Completed None 13 25 55 389 | INFO | Task run my mapped task ca1f3dbb 3 Finished in state Completed None 13 25 57 393 | INFO | Task run my mapped task ca1f3dbb 4 Finished in state Completed None 13 25 59 413 | INFO | Task run my mapped task ca1f3dbb 5 Finished in state Completed None 13 26 01 417 | INFO | Task run my mapped task ca1f3dbb 6 Finished in state Completed None 13 26 03 425 | INFO | Task run my mapped task ca1f3dbb 7 Finished in state Completed None 13 26 05 441 | INFO | Task run my mapped task ca1f3dbb 8 Finished in state Completed None 13 26 07 586 | INFO | Task run my mapped task ca1f3dbb 9 Finished in state Completed None distributed dask worker INFO Exiting on signal 15```

Thomas Mignon · Answer

But there is no sign of the prints

Thomas Mignon · Answer

And if i try to split my hello world code into 2 file in order to have a flow calling a flow calling a task who looks what i m trying to achieve with my real code who is broken by now but it s better to do it with a hello world first i have hello world py who is ```from prefect import flow from prefect graph hello world task import hello world task flow flow def hello world flow for i in range 10 hello world task flow i if name == main hello world flow ``` and hello world task py who is ```from prefect import flow task from prefect task runners import DaskTaskRunner flow task runner=DaskTaskRunner cluster class= dask jobqueue PBSCluster cluster kwargs= cores 1 memory 24GB queue sequentiel walltime 00 05 00 interface ib0 n workers 1 def hello world task flow x my mapped task x task def my mapped task x print x ``` I got the same error in the dask worker o a timeout error for the 10 flow that are called ```distributed nanny INFO Start Nanny at distributed worker INFO Start worker at distributed worker INFO Listening to distributed worker INFO dashboard at 10 148 1 114 44616 distributed worker INFO Waiting to connect to distributed worker INFO distributed worker INFO Threads 1 distributed worker INFO Memory 22 35 GiB distributed worker INFO Local Directory dev shm pbs 9835069 datarmor0 dask worker space worker 0oganzzi distributed worker INFO distributed worker INFO Registered to distributed worker INFO distributed core INFO Starting established connection distributed core INFO Event loop was unresponsive in Worker for 12 55s This is often caused by long running GIL holding functions or moving large chunks of data This can cause timeouts and instability distributed worker WARNING Compute Failed Function begin task run args kwargs task lt prefect tasks Task object at 0x2aaab091ed90 gt task run TaskRun id=UUID bd7c31c6 f577 49ee a438 f7db41ec1069 name= my mapped task 60f685bd 2 flow run id=UUID 9b7aee9e 6453 4e36 8518 cedfa012f375 task key= 60f685bd73a2d907cce1732888a4657e dynamic key= 2 cache key=None cache expiration=None task version=None empirical policy=TaskRunPolicy max retries=0 retry delay seconds=0 0 tags= state id=UUID 63e4ad1a d7e4 4b33 8631 ac205202d59d task inputs= x state type=StateType PENDING run count=0 expected start time=datetime datetime 2022 4 6 13 23 6 309348 tzinfo=datetime timezone utc next scheduled start time=None start time=None end time=None total run time=datetime timedelta 0 estimated run time=datetime timedelta 0 estimated start time delta=datetime timedelta microseconds=61193 state=Pending message=None type=PENDING result=None task run id=bd7c31c6 f577 49ee a438 f7db41ec1069 parameters x 2 wait for None Exception TimeoutError Terminated distributed dask worker INFO Exiting on signal 15 distributed nanny INFO Closing Nanny at distributed nanny INFO Worker process 26133 was killed by signal 15 distributed dask worker INFO End worker```

Anna Geller · Answer

instead of prints you need to use logs you can use `get run logger` ```from prefect import get run logger task flow task def compute something x logger = get run logger <http logger info|logger info> Computing d x 2 = d x x 2 time sleep 2 ```

Thomas Mignon · Answer

Ok i will try with my 2 versions of the hello world to see whats going on slightly smiling face

Thomas Mignon · Answer

The version with one file is ok i see the log in my dask worker output file

Thomas Mignon · Answer

But for the other version this is ok too i see the logs when the task of the flow of a flow are launched but it crash at the third fourth fifth sometime i don t know why

Thomas Mignon · Answer

here is what i see as error this time this is at the second

Thomas Mignon · Answer

```13 55 00 356 | INFO | prefect engine Created flow run rigorous hog for flow hello world flow 13 55 00 356 | INFO | Flow run rigorous hog Using task runner ConcurrentTaskRunner 13 55 01 420 | INFO | Flow run rigorous hog Created subflow run spectral goldfish for flow hello world task flow 13 55 01 422 | INFO | prefect task runner dask Creating a new Dask cluster with `dask jobqueue pbs PBSCluster` 13 55 01 989 | INFO | prefect task runner dask The Dask dashboard is available at 13 55 02 960 | INFO | Flow run spectral goldfish Created task run my mapped task 60f685bd 0 for task my mapped task home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect client py 1195 UserWarning No default storage has been set on the server Using temporary local storage for results warnings warn 13 55 29 286 | INFO | Flow run spectral goldfish Finished in state Completed All states completed 13 55 30 308 | INFO | Flow run rigorous hog Created subflow run pink bulldog for flow hello world task flow 13 55 30 310 | INFO | prefect task runner dask Creating a new Dask cluster with `dask jobqueue pbs PBSCluster` 13 55 30 753 | INFO | prefect task runner dask The Dask dashboard is available at 13 55 31 428 | INFO | Flow run pink bulldog Created task run my mapped task 60f685bd 1 for task my mapped task home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect client py 1195 UserWarning No default storage has been set on the server Using temporary local storage for results warnings warn 13 56 02 864 | INFO | Flow run pink bulldog Finished in state Completed All states completed 13 56 04 668 | INFO | Flow run rigorous hog Created subflow run offbeat mamba for flow hello world task flow 13 56 04 670 | INFO | prefect task runner dask Creating a new Dask cluster with `dask jobqueue pbs PBSCluster` 13 56 04 975 | INFO | prefect task runner dask The Dask dashboard is available at 13 56 06 351 | INFO | Flow run offbeat mamba Created task run my mapped task 60f685bd 2 for task my mapped task home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect client py 1195 UserWarning No default storage has been set on the server Using temporary local storage for results warnings warn 13 56 27 540 | ERROR | Flow run offbeat mamba Crash detected Execution was interrupted by an unexpected exception 13 56 27 965 | ERROR | Flow run rigorous hog Encountered exception during execution Traceback most recent call last File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect engine py line 456 in orchestrate flow run result = await run sync in worker thread flow call File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect utilities asyncio py line 52 in run sync in worker thread return await anyio to thread run sync context run call cancellable=True File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages anyio to thread py line 28 in run sync return await get asynclib run sync in worker thread func args cancellable=cancellable File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages anyio backends asyncio py line 818 in run sync in worker thread return await future File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages anyio backends asyncio py line 754 in run result = context run func args File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect graph hello world py line 9 in hello world flow hello world task flow i File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect flows py line 314 in call return enter flow run engine from flow call self parameters File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect engine py line 114 in enter flow run engine from flow call return run async from worker thread begin run File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect utilities asyncio py line 63 in run async from worker thread return anyio from thread run call File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages anyio from thread py line 35 in run return asynclib run async from thread func args File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages anyio backends asyncio py line 847 in run async from thread return f result File home datawork semaphore exp conda conda envs prefect lib python3 9 concurrent futures base py line 445 in result return self get result File home datawork semaphore exp conda conda envs prefect lib python3 9 concurrent futures base py line 390 in get result raise self exception File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect client py line 82 in with injected client return await fn args kwargs File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect engine py line 369 in create and begin subflow run terminal state = await orchestrate flow run File home datawork semaphore exp conda conda envs prefect lib python3 9 contextlib py line 670 in aexit raise exc details 1 File home datawork semaphore exp conda conda envs prefect lib python3 9 contextlib py line 199 in aexit await self gen athrow typ value traceback File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect engine py line 866 in report flow run crashes raise exc from None File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect engine py line 849 in report flow run crashes yield File home datawork semaphore exp conda conda envs prefect lib python3 9 contextlib py line 653 in aexit cb suppress = await cb exc details File home datawork semaphore exp conda conda envs prefect lib python3 9 contextlib py line 188 in aexit await self gen anext File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect task runners py line 158 in start self started = False File home datawork semaphore exp conda conda envs prefect lib python3 9 contextlib py line 670 in aexit raise exc details 1 File home datawork semaphore exp conda conda envs prefect lib python3 9 contextlib py line 653 in aexit cb suppress = await cb exc details File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages distributed deploy cluster py line 474 in aexit await f File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages distributed deploy spec py line 414 in close assert w status == Status closed w status AssertionError Status running home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect client py 1195 UserWarning No default storage has been set on the server Using temporary local storage for results warnings warn 13 56 28 924 | ERROR | Flow run rigorous hog Finished in state Failed Flow run encountered an exception ```

Thomas Mignon · Answer

and if i look inside the qsub log of this tasks she hasn t crashed in reality ```Thus no job control in this shell distributed nanny INFO Start Nanny at distributed worker INFO Start worker at distributed worker INFO Listening to distributed worker INFO dashboard at 10 148 0 171 52201 distributed worker INFO Waiting to connect to distributed worker INFO distributed worker INFO Threads 1 distributed worker INFO Memory 22 35 GiB distributed worker INFO Local Directory dev shm pbs 9835312 datarmor0 dask worker space worker q7ltl636 distributed worker INFO distributed worker INFO Registered to distributed worker INFO distributed core INFO Starting established connection distributed core INFO Event loop was unresponsive in Worker for 9 00s This is often caused by long running GIL holding functions or moving large chunks of data This can cause timeouts and instability 13 56 24 283 | INFO | Task run my mapped task 60f685bd 2 This is x 2 home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect client py 1195 UserWarning No default storage has been set on the server Using temporary local storage for results warnings warn 13 56 26 897 | INFO | Task run my mapped task 60f685bd 2 Finished in state Completed None distributed dask worker INFO Exiting on signal 15 distributed nanny INFO Closing Nanny at Terminated distributed nanny INFO Worker process 925 was killed by signal 15 distributed dask worker INFO End worker```

Thomas Mignon · Answer

Seems strange

Anna Geller · Answer

re this error ```UserWarning No default storage has been set on the server Using temporary local storage for results ``` you can create and set default storage using ```prefect storage create```

Thomas Mignon · Answer

Thank you for all the help and this tips warning gone slightly smiling face I m searching now why my flow of tasks crash randomly on a task with a timeout when other ones are ok maybe it come from the queue sequentiel i m using on the pbs cluster or from the config or something else

Thomas Mignon · Answer

I ve tried a lot of conf one seems to be better but only when i m doing a flow of 10 task so 1 qsub to PBS cluster But this config doesn t work when i m doing a flow of 10 flow who call a task so 10 qsub behind to my PBS cluster cf this trace ```07 34 03 307 | INFO | prefect engine Created flow run armored whale for flow hello world flow 07 34 03 308 | INFO | Flow run armored whale Using task runner ConcurrentTaskRunner 07 34 04 387 | INFO | Flow run armored whale Created subflow run shrewd marmot for flow hello world task flow 07 34 04 389 | INFO | prefect task runner dask Creating a new Dask cluster with `dask jobqueue pbs PBSCluster` 07 34 04 615 | INFO | prefect task runner dask The Dask dashboard is available at 07 34 05 165 | INFO | Flow run shrewd marmot Created task run my mapped task 60f685bd 0 for task my mapped task 07 34 17 860 | INFO | Flow run shrewd marmot Finished in state Completed All states completed 07 34 18 871 | INFO | Flow run armored whale Created subflow run nifty lizard for flow hello world task flow 07 34 18 873 | INFO | prefect task runner dask Creating a new Dask cluster with `dask jobqueue pbs PBSCluster` 07 34 18 949 | INFO | prefect task runner dask The Dask dashboard is available at 07 34 19 497 | INFO | Flow run nifty lizard Created task run my mapped task 60f685bd 1 for task my mapped task 07 34 45 914 | INFO | Flow run nifty lizard Finished in state Completed All states completed 07 34 46 799 | INFO | Flow run armored whale Created subflow run ambitious sponge for flow hello world task flow 07 34 46 801 | INFO | prefect task runner dask Creating a new Dask cluster with `dask jobqueue pbs PBSCluster` 07 34 47 043 | INFO | prefect task runner dask The Dask dashboard is available at 07 34 47 589 | INFO | Flow run ambitious sponge Created task run my mapped task 60f685bd 2 for task my mapped task 07 35 01 606 | ERROR | Flow run ambitious sponge Crash detected Execution was interrupted by an unexpected exception ``` So i m trying to put a retries mechanism on my task but it doesn t work here is what i do and nothing change no retry is tried ``` flow task runner=DaskTaskRunner cluster class= dask jobqueue PBSCluster cluster kwargs= cores 1 memory 24GB queue mpi 4 resource spec select=4 ncpus=28 mem=120gb walltime 00 05 00 interface ib0 n workers 1 def hello world task flow x my mapped task x task retries=10 retry delay seconds=5 def my mapped task x logger = get run logger This is x d x time sleep 2 ``` Maybe i have to put a retry on the flow too which doesn t seems to be something do able in prefect Because maybe it s not the task itself that crash but the flow when it does the qsub to PBS so when it crash we never reach the task because crash occurs in the flow

Anna Geller · Answer

I m no HPC expert and I didn t understand your issue Can you explain in simple terms what problem do you see here So far I see no mapping no parallelizable operation within `my mapped task` Not much for PBSCluster to do Maybe you can try this again which I remember worked ```from prefect import flow task from prefect task runners import DaskTaskRunner task def my mapped task x print x flow task runner=DaskTaskRunner cluster class= dask jobqueue PBSCluster cluster kwargs= cores 1 memory 24GB queue sequentiel walltime 00 05 00 interface ib0 n workers 1 def hello world flow for i in range 10 my mapped task i if name == main hello world flow ``` looks like you changed the queue and resource spec as compared to the last time so maybe this can be the issue

Thomas Mignon · Answer

The example you paste work but it does one qsub to the HPC cluster then start a dask worker inside the qsub and then launch the 10 task inside the dask worker The thing i want to test now is to do 10 qsub in order to launch 1 task in each to the HPC i don t know if i m clear

Thomas Mignon · Answer

In order to do that i ve split my code in two file with two flows you can forgot the queue and ressource spec this is not the problem slightly smiling face ``` flow task runner=DaskTaskRunner cluster class= dask jobqueue PBSCluster cluster kwargs= cores 1 memory 24GB queue sequentiel walltime 00 50 00 interface ib0 n workers 1 def hello world task flow x my mapped task x task retries=10 retry delay seconds=5 def my mapped task x logger = get run logger <http logger info|logger info> This is x d x time sleep 2 ``` and ```from prefect import flow from prefect graph hello world task import hello world task flow flow def hello world flow for i in range 10 hello world task flow i if name == main hello world flow ``` But it crash randomly between two task and the last task was successfully done so i m thinking that s when the new qsub is done there is a problem and the qsub is submitted on the flow that s why i would try to put a retry on the flow not on the task

Thomas Mignon · Answer

I don t know if my explication is better sweat smile

Anna Geller · Answer

we don t have flow level retries as of now but this is in the backlog already and may be added in the future re your issue I m not sure whether it makes sense to submit 1 task flows to your PBS cluster this way Are you just doing it to explore various options or is there some concrete problem you are trying to solve this way moving the loop to the PBS DaskTaskRunner flow makes more sense ```def hello world task flow x for i in range 10 my mapped task i ```

Thomas Mignon · Answer

I m doing it this way because the graph that i need to produce for my project gonna be a flow of subflows with many tasks in each subflows more than 1 but it s to simplify the helloworld

Thomas Mignon · Answer

I think flow subflow level retries could be a good solution but if i can found a temporary solution to my problem it would be so great slightly smiling face

Anna Geller · Answer

I see but do you believe you need to map over flows you could map parallelize your data flow operations in one flow and then use some parent flow just to call those subflows not necessarily via mapping right It all depends on your use case

Thomas Mignon · Answer

I don t understand the question the for is a map for you I don t need this for map in my real use case but i will need to have one parent flow that has subflows with tasks so i can rewrite my hello world with subflow like this in order to be close to my real use case is this what you mean to when saying parallelize ```from prefect import flow from prefect graph hello world task import hello world task flow flow def hello world flow hello world task flow 1 hello world task flow 2 hello world task flow 3 hello world task flow 4 hello world task flow 5 hello world task flow 6 hello world task flow 7 hello world task flow 8 hello world task flow 9 hello world task flow 10 if name == main hello world flow ```

Anna Geller · Answer

yup exactly for loop is the same as mapping = looping over some function with your business logic it may be a task or subflow by parallelized I meant running tasks in parallel

Anna Geller · Answer

my main point is each of your child flows that needs parallelism may use either Dask or Ray or just a `CuncurrentTaskRunner` for parallelism and you can decide on which one to use based on the business logic of each subflow and then your main flow can just trigger the subflows in the right order but you don t necessarily need to map over subflows your use case with passing an incremental number to each subflow seem unlikely in practice would you agree slightly smiling face a more likely scenario would be something like this only subflow 2 doing something where PBS Cluster may be helpful in some way ``` flow def process ml workflow subflow 1 is doing IO operation so it may be more efficient to use ConcurrentTaskRunner rather than PBS DaskCluster subflow 1 = extract raw data concurrently and load to some destination subflow 2 = read and process data in parallel wait for= subflow 1 this may use PBS subflow 3 = train some ml model and pickle it wait for= subflow 2 ```

Thomas Mignon · Answer

Yes i agree slightly smiling face I will take your version and adapt it to my real use case thank you a lot for all your help during these past days you was very helpful and you explain the things in a very simple way