https://prefect.io logo
Join Slack
Powered by
hey everyone - im dealing with an unusual issue in...
# ask-community
j
hey everyone - im dealing with an unusual issue in dask. I need to clear memory after a workflow run (success or failure) but I can’t use 
client.restart
in the workflow because the workflow will fail. Any thoughts or ideas on how to clear memory
c
Hi @John Ramirez - you can call 
client.restart
in the 
on_exit
callback of your Flow’s execution environment: https://docs.prefect.io/cloud/execution/overview.html#environments
j
so i found that but i have my dask cluster within EKS which is not open to the public. when I try to deploy my workflow via docker, the function is validated and breaks
The function also run at deployment which I also do not want
c
I don’t think I understand -> if your workflow is successfully running against your Dask cluster, then it must have access to that cluster so this command should work
j
yes i do have access to the cluster but I use the kubernetes internal IP address in the workflow not the external address
c
Sure, but regardless if you can access the cluster then you can call restart whenever the workflow has completed its run via the on exit hook. Did I maybe misunderstand the goal?
j
no you are right but i dont understand why the function is executed during deployment. Is there a way to turn that off
c
I’m sorry I don’t think I understand your question; the way Cloud deployments on K8s work at a high level is: Agent creates K8s job -> Environment class spins up and calls setup, execute, Flow runs and submits work to whatever executor, on exit is called, job is cleaned up
j
sorry i meant when I run 
flow.register()
the 
on_start
or 
on_exit
function also runs
c
Hmm that shouldn’t happen
j
do want an error log
c
Yea, that would be great
j
Copy code
Traceback (most recent call last):
  File "/Users/johnramirez/.local/share/virtualenvs/trader-ranking-Oomc_S11/lib/python3.7/site-packages/distributed/comm/core.py", line 218, in connect
    _raise(error)
  File "/Users/johnramirez/.local/share/virtualenvs/trader-ranking-Oomc_S11/lib/python3.7/site-packages/distributed/comm/core.py", line 203, in _raise
    raise IOError(msg)
OSError: Timed out trying to connect to '<tcp://dask-scheduler.default.svc.cluster.local:8786>' after 10 s: connect() didn't finish in time

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "main.py", line 92, in <module>
    on_start=clear_memory()
  File "main.py", line 79, in clear_memory
    client = Client('dask-scheduler.default.svc.cluster.local:8786')
  File "/Users/johnramirez/.local/share/virtualenvs/trader-ranking-Oomc_S11/lib/python3.7/site-packages/distributed/client.py", line 723, in __init__
    self.start(timeout=timeout)
  File "/Users/johnramirez/.local/share/virtualenvs/trader-ranking-Oomc_S11/lib/python3.7/site-packages/distributed/client.py", line 896, in start
    sync(self.loop, self._start, **kwargs)
  File "/Users/johnramirez/.local/share/virtualenvs/trader-ranking-Oomc_S11/lib/python3.7/site-packages/distributed/utils.py", line 348, in sync
    raise exc.with_traceback(tb)
  File "/Users/johnramirez/.local/share/virtualenvs/trader-ranking-Oomc_S11/lib/python3.7/site-packages/distributed/utils.py", line 332, in f
    result[0] = yield future
  File "/Users/johnramirez/.local/share/virtualenvs/trader-ranking-Oomc_S11/lib/python3.7/site-packages/tornado/gen.py", line 735, in run
    value = future.result()
  File "/Users/johnramirez/.local/share/virtualenvs/trader-ranking-Oomc_S11/lib/python3.7/site-packages/distributed/client.py", line 991, in _start
    await self._ensure_connected(timeout=timeout)
  File "/Users/johnramirez/.local/share/virtualenvs/trader-ranking-Oomc_S11/lib/python3.7/site-packages/distributed/client.py", line 1048, in _ensure_connected
    connection_args=self.connection_args,
  File "/Users/johnramirez/.local/share/virtualenvs/trader-ranking-Oomc_S11/lib/python3.7/site-packages/distributed/comm/core.py", line 227, in connect
    _raise(error)
  File "/Users/johnramirez/.local/share/virtualenvs/trader-ranking-Oomc_S11/lib/python3.7/site-packages/distributed/comm/core.py", line 203, in _raise
    raise IOError(msg)
OSError: Timed out trying to connect to '<tcp://dask-scheduler.default.svc.cluster.local:8786>' after 10 s: Timed out trying to connect to '<tcp://dask-scheduler.default.svc.cluster.local:8786>' after 10 s: connect() didn't finish in time
here is the python code as well
Copy code
def clear_memory():
    client = Client('dask-scheduler.default.svc.cluster.local:8786')
    client.restart()

ENV = RemoteEnvironment(
    executor="prefect.engine.executors.DaskExecutor",
    executor_kwargs={
        "address": "<tcp://dask-scheduler.default.svc.cluster.local:8786>"
    },
    on_start=clear_memory()
)
c
You are running your 
clear_memory
function in this code; you should instead be providing that function to `on_start`:
Copy code
ENV = RemoteEnvironment(
    executor="prefect.engine.executors.DaskExecutor",
    executor_kwargs={
        "address": "<tcp://dask-scheduler.default.svc.cluster.local:8786>"
    },
    on_start=clear_memory # <--- the change I made
)
j
thank you that worked.
c
👍 np
2 Views
Open in Slack
PreviousNext
Powered by