Hey all I have a Prefect 2 flow using DaskTaskRunner Reading

Question

Hey all I have a Prefect 2 flow using DaskTaskRunner Reading data from multiple unrelated sources preprocessing and then training a model I also use map as part of the flow for parsing some objects in parallel Everything works fine until I get this error related to Dask ```12 36 08 900 | INFO | Flow run logical peacock Created task run calculate additional features 0 for task calculate additional features 2023 04 30 12 36 08 915 distributed protocol pickle ERROR Failed to serialize lt ToPickle HighLevelGraph with 1 layers ``` What is the problem here what am I missing Sharing my flow code as a comment on this message Setup running locally with pycharm s docker compose interpeter Prefect 2 10 4 prefect dask==0 2 3 Thanks

Dekel R · Answer

< Ofek Katriel> FYI ```from prefect import flow get run logger unmapped from src transformatio tasks import get asins to transform load model embedding df check existing transformed asins get pdp data load nth recent file from storage get titles without brand and unit words sentence preprocess calculate additional features from alerts import send slack alert from filter layer logic import save to partiotions from src adp tasks import missing asins request from math import ceil import pandas as pd from datetime import datetime from src configurations import PRODUCTION BUCKET TF IDF FOLDER TRANSFORM BATCH SIZE from prefect dask task runners import DaskTaskRunner flow name= similarity filter layer description= Preprocessing for similarity filter layer retries=2 task runner=DaskTaskRunner def flow is onboarding bool = False onboarding asins path str = None logger = get run logger Starting filter layer preprocessing missing asins request submit required model asins = load model embedding df submit existing transformed asins = check existing transformed asins submit asins to check=required model asins asins to transform = get asins to transform submit required model asins=required model asins existing transformed asins=existing transformed asins Starting transform process pdp df = get pdp data submit asins to transform = asins to transform result tf idf dict = load nth recent file from storage submit bucket=PRODUCTION BUCKET folder=TF IDF FOLDER file rank=1 total batches = ceil len pdp df TRANSFORM BATCH SIZE total batches = 1 transform df list = f Starting transform per batch loop for i in range total batches pdp slice = pdp df iloc i TRANSFORM BATCH SIZE i + 1 TRANSFORM BATCH SIZE df title without brand list unit words = get titles without brand and unit words submit input df=pdp slice result processed title list = sentence preprocess map unit words=unmapped unit words is stem=unmapped True sentence=title without brand list processed title df = pd DataFrame df asin to list columns= asin processed title df title processed stem = pd Series processed title list transform df = calculate additional features submit df=pdp slice tf idf dict=tf idf dict tf idf partition var= cat 0 processed title df=processed title df result f Transform batch i + 1 total batches done transform df list append transform df del transform df f Finished transforming full transformed df = pd concat transform df list full transformed df transform ts = datetime now saved to partitions = save to partiotions submit wait for=full transformed df saved to partitions state = saved to partitions wait if saved to partitions state is failed The filter layer flow failed Look at the logs send slack alert The filter layer flow failed Look at the logs else The filter layer flow finished running send slack alert The filter layer flow finished running if name == main flow ``` I get the error in this line ```transform df = calculate additional features submit df=pdp slice tf idf dict=tf idf dict tf idf partition var= cat 0 processed title df=processed title df result ```

Dekel R · Answer

I solvedd this byy getting some of the nativePythonn code inside a task instead of using it in the flow context Specifically the following lines ```processed title df = pd DataFrame df asin to list columns= asin processed title df title processed stem = pd Series processed title list ```

Dekel R · Answer

After running the flow locally I deployed it and got this error after the map function I m using see the error ```Crash detected Execution was interrupted by an unexpected exception Traceback most recent call last File opt conda lib python3 10 site packages h2 connection py line 224 in process input func target state = self transitions self state input KeyError lt ConnectionState CLOSED 3 gt lt ConnectionInputs SEND HEADERS 0 gt During handling of the above exception another exception occurred Traceback most recent call last File opt conda lib python3 10 site packages httpcore async http2 py line 116 in handle async request await self send request headers request=request stream id=stream id File opt conda lib python3 10 site packages httpcore async http2 py line 213 in send request headers self h2 state send headers stream id headers end stream=end stream File opt conda lib python3 10 site packages h2 connection py line 766 in send headers self state machine process input ConnectionInputs SEND HEADERS File opt conda lib python3 10 site packages h2 connection py line 228 in process input raise ProtocolError h2 exceptions ProtocolError Invalid input ConnectionInputs SEND HEADERS in state ConnectionState CLOSED During handling of the above exception another exception occurred Traceback most recent call last File opt conda lib python3 10 site packages httpx transports default py line 60 in map httpcore exceptions yield File opt conda lib python3 10 site packages httpx transports default py line 353 in handle async request resp = await self pool handle async request req File opt conda lib python3 10 site packages httpcore async connection pool py line 253 in handle async request raise exc File opt conda lib python3 10 site packages httpcore async connection pool py line 237 in handle async request response = await connection handle async request request File opt conda lib python3 10 site packages httpcore async connection py line 90 in handle async request return await self connection handle async request request File opt conda lib python3 10 site packages httpcore async http2 py line 152 in handle async request raise LocalProtocolError exc pragma nocover httpcore LocalProtocolError Invalid input ConnectionInputs SEND HEADERS in state ConnectionState CLOSED The above exception was the direct cause of the following exception httpx LocalProtocolError Invalid input ConnectionInputs SEND HEADERS in state ConnectionState CLOSED 02 59 38 PM prefect flow runs``` Any suggestions Thanks

Luis Cebrián · Answer

I got the same error today This helped me and specifically comment

Luis Cebrián · Answer

Regarding your Dask error I m afraid I have no dask experience to help you

Dekel R · Answer

Thanks I did it and now I get this error ```Encountered exception during execution Traceback most recent call last File opt conda lib python3 10 site packages httpx transports default py line 60 in map httpcore exceptions yield File opt conda lib python3 10 site packages httpx transports default py line 353 in handle async request resp = await self pool handle async request req File opt conda lib python3 10 site packages httpcore async connection pool py line 253 in handle async request raise exc File opt conda lib python3 10 site packages httpcore async connection pool py line 237 in handle async request response = await connection handle async request request File opt conda lib python3 10 site packages httpcore async connection py line 86 in handle async request raise exc File opt conda lib python3 10 site packages httpcore async connection py line 63 in handle async request stream = await self connect request File opt conda lib python3 10 site packages httpcore async connection py line 150 in connect stream = await stream start tls kwargs File opt conda lib python3 10 site packages httpcore backends asyncio py line 66 in start tls with map exceptions exc map File opt conda lib python3 10 contextlib py line 153 in exit self gen throw typ value traceback File opt conda lib python3 10 site packages httpcore exceptions py line 14 in map exceptions raise to exc exc httpcore ConnectError The above exception was the direct cause of the following exception Traceback most recent call last File opt conda lib python3 10 site packages prefect engine py line 669 in orchestrate flow run result = await flow call aresult File opt conda lib python3 10 site packages prefect internal concurrency calls py line 181 in aresult return await asyncio wrap future self future File opt conda lib python3 10 site packages prefect internal concurrency calls py line 194 in run sync result = self fn self args self kwargs File app flow py line 59 in flow tf idf partition var= cat 0 result File opt conda lib python3 10 site packages prefect futures py line 232 in result result File opt conda lib python3 10 site packages prefect internal concurrency calls py line 173 in result return self future result timeout=timeout File opt conda lib python3 10 concurrent futures base py line 458 in result return self get result File opt conda lib python3 10 concurrent futures base py line 403 in get result raise self exception File opt conda lib python3 10 site packages prefect internal concurrency calls py line 218 in run async result = await coro File opt conda lib python3 10 site packages prefect futures py line 241 in result return await final state result raise on failure=raise on failure fetch=True File opt conda lib python3 10 site packages prefect states py line 91 in get state result raise await get state exception state File opt conda lib python3 10 site packages prefect dask task runners py line 271 in wait return await future result timeout=timeout File opt conda lib python3 10 site packages distributed client py line 329 in result raise exc with traceback tb File opt conda lib python3 10 site packages prefect engine py line 1370 in begin task run state = await orchestrate task run File opt conda lib python3 10 site packages prefect engine py line 1447 in orchestrate task run flow run = await client read flow run task run flow run id File opt conda lib python3 10 site packages prefect client orchestration py line 1609 in read flow run response = await self client get f flow runs flow run id File opt conda lib python3 10 site packages httpx client py line 1754 in get return await self request File opt conda lib python3 10 site packages httpx client py line 1530 in request return await self send request auth=auth follow redirects=follow redirects File opt conda lib python3 10 site packages prefect client base py line 246 in send response = await self send with retry File opt conda lib python3 10 site packages prefect client base py line 192 in send with retry response = await request File opt conda lib python3 10 site packages httpx client py line 1617 in send response = await self send handling auth File opt conda lib python3 10 site packages httpx client py line 1645 in send handling auth response = await self send handling redirects File opt conda lib python3 10 site packages httpx client py line 1682 in send handling redirects response = await self send single request request File opt conda lib python3 10 site packages httpx client py line 1719 in send single request response = await transport handle async request request File opt conda lib python3 10 site packages httpx transports default py line 352 in handle async request with map httpcore exceptions File opt conda lib python3 10 contextlib py line 153 in exit self gen throw typ value traceback File opt conda lib python3 10 site packages httpx transports default py line 77 in map httpcore exceptions raise mapped exc message from exc httpx ConnectError 12 12 12 PM prefect flow runs Crash detected Request to <https api prefect cloud api accounts 0aceff1b 4fdb 4469 adde 197148a98672 workspaces b7474c31 4033 449b 96b6 f3544e6f9408 flow runs 8907a295 5335 4262 99fe a6921dbba067> failed Traceback most recent call last File opt conda lib python3 10 site packages anyio streams tls py line 130 in call sslobject method result = func args File opt conda lib python3 10 ssl py line 975 in do handshake self sslobj do handshake ssl SSLSyscallError Some I O error occurred ssl c 997 The above exception was the direct cause of the following exception Traceback most recent call last File opt conda lib python3 10 site packages httpcore exceptions py line 10 in map exceptions yield File opt conda lib python3 10 site packages httpcore backends asyncio py line 78 in start tls raise exc File opt conda lib python3 10 site packages httpcore backends asyncio py line 69 in start tls ssl stream = await anyio streams tls TLSStream wrap File opt conda lib python3 10 site packages anyio streams tls py line 122 in wrap await wrapper call sslobject method ssl object do handshake File opt conda lib python3 10 site packages anyio streams tls py line 151 in call sslobject method raise BrokenResourceError from exc anyio BrokenResourceError During handling of the above exception another exception occurred Traceback most recent call last File opt conda lib python3 10 site packages httpx transports default py line 60 in map httpcore exceptions yield File opt conda lib python3 10 site packages httpx transports default py line 353 in handle async request resp = await self pool handle async request req File opt conda lib python3 10 site packages httpcore async connection pool py line 253 in handle async request raise exc File opt conda lib python3 10 site packages httpcore async connection pool py line 237 in handle async request response = await connection handle async request request File opt conda lib python3 10 site packages httpcore async connection py line 86 in handle async request raise exc File opt conda lib python3 10 site packages httpcore async connection py line 63 in handle async request stream = await self connect request File opt conda lib python3 10 site packages httpcore async connection py line 150 in connect stream = await stream start tls kwargs File opt conda lib python3 10 site packages httpcore backends asyncio py line 66 in start tls with map exceptions exc map File opt conda lib python3 10 contextlib py line 153 in exit self gen throw typ value traceback File opt conda lib python3 10 site packages httpcore exceptions py line 14 in map exceptions raise to exc exc httpcore ConnectError The above exception was the direct cause of the following exception httpx ConnectError ```