Closed ahuang11 closed 1 month ago
@ahuang11, @madkinsz hello, Did you resolve this issue?
I have met the same issue :(
And it seems to be a little bit random, sometimes my flows are executed without this problem.
I have same iusse in RayTaskRunner. :(
For me, the situation has been consistently reproduced in the following situation:
@task(tags={'load'}, log_prints=True, retries=3, retry_delay_seconds=5,)
async def load_raw(url:str, auth:Auth|None=None, **kwds) -> bytes:
content = await _load_raw_inner(url, auth=auth, **kwds)
return content
_load_raw = load_raw.with_options(
name=f'load-raw-{PROV_NAME.lower()}', tags=load_raw.tags | {PROV_NAME},
cache_key_fn=task_input_hash, persist_result=True,
cache_expiration=_cache_expiration, retries=3, retry_delay_seconds=5,)
@flow
async def epool_rem_gr_active_group_csv():
cont = await _load_raw(URL, verify=False, timeout=15.)
...
Fixed it by moving the caching parameters to the task definition:
@task(
tags={'load'}, log_prints=True,
cache_key_fn=task_input_hash, persist_result=True, retries=3, retry_delay_seconds=5,)
async def load_raw(url:str, auth:Auth|None=None, **kwds) -> bytes:
content = await _load_raw_inner(url, auth=auth, **kwds)
return content
_load_raw = load_raw.with_options(
name=f'load-raw-{PROV_NAME.lower()}', tags=load_raw.tags | {PROV_NAME},
cache_expiration=_cache_expiration,)
@flow
async def epool_rem_gr_active_group_csv():
cont = await _load_raw(URL, verify=False, timeout=15.)
...
Fix do not work. :(
see also #8228
and #8415
Downgrate prefect to 2.16.9 The error has almost disappeared
This pattern of result persistence is updated and fixed in 3.0 - I'm going to close this but if there are other issues that arise, please open a new issue.
Occurs when a worker's memory overflows (with_get_dask_client is commented out)
Logs:
Traceback: