https://prefect.io logo
Join Slack
Powered by
I'm seeing variables change within a mapped task w...
# ask-community
n
I'm seeing variables change within a mapped task where I have the task call a function. The function prints out the row mapped by the dataframe and returns that value, but later uses of the variable have different values than what was set by the function previously. Did something change from v2 to v3? Should I make the function a task to have it behave normally?
n
can you show a minimal example of what you mean? it sounds like the implementation of your task is mutating an object in place unexpectedly? there shouldn't be any differences from 2 to 3 that would affect this
n
I'll see if I can get a minimal example of what I mean, it's a strange one. Mapping over a dataframe, use a function to get values out of the dataframe, and the values change over time after the function has run.
HI Nate, ended up figuring out what happened, was passing an unmapped list through to a task I was mapping, which seemed to work ok (guessing the task generator was making copies) in v2, but didn't in v3. Ended up making copies of the list in the task itself. Unsure if passing along a minimal example is going to be useful as I'm guessing this is a user error thing.
n
i’d suggest it’d still be useful as if it’s truly something that worked in 2 and not in 3 then we should add it to this page
happy to put a PR up for that if you have an example to share
n
Makes sense to me, I'll still send the minimal code over. Just have to get it off the company computer and onto a personal as the company doesn't allow slack. Another thing that's happening is that the mapping of three tasks in a row seems to take around 1-3 minutes in 3 vs instantanously in v2. Anything I should look out for that leads to that? chatgpt says I needed to materialize the data frame I'm passing into the first task, but didn't make much difference. I'm also seeing tasks that take a second continue infinitely on the ui. I'll get some examples off the work computer
n
great thanks! i think seeing the code will make it lot easier to understand the behavior differences you're seeing here the only bi differences here between v2 and v3 is that in v3 you must call 
.result()
or 
.wait()
on terminal futures (futures you don't pass to another task / flow or 
return
) or else you're not guaranteeing that those futures are not garbage collected but it sounds like that might not be what you're running into. ill keep on the lookout for your examples
n
Probably the best I can do at the moment in terms of code differences. I can probably supply the code in text if that would help. You can see the unmapped(geneva_args) list I was passing to a task I was mapping that was getting changed in the task and affecting other versions of the task as well.
geneva extract v2 vs v3.docx
Here's the code to reproduce the issue I was seeing with values changing over tasks which seemed to work ok in v2 but not in v3. Ended up fixing it by doing a dict/list copy in the task before making changes.
minimal_weird_bug_reproduce.py.txt
n
i appreciate it, thank you! will take a look
n
Sure thing, the lines where I comment with a <- are probably the most helpful. Took a bit of digging to see which variable was the one getting switched all over the place, so hopefully the code isn't too verbose.
n
n
Thanks, any ideas on what could be causing the time difference between creating mapped tasks in v2 and v3?
n
in general creating mapped tasks in 3 is much faster than 2. so if that's not the case i suspect something funny/unexpected is going on happy to look at an MRE where that's the case (as in like no company / bespoke proper nouns) otherwise i can come back to this later and ask claude to boil your code down to an MRE
n
Thanks, yeah it’s a tough one to reproduce. Will work on the other mre I sent you earlier and see if I can make the same thing happen. Does passing data frames between tasks affect task creation speed?
n
👍
for large objects like big dfs yea we have to recursively check it for futures you might not have resolved so you can 
quote
it but then we wont check for futures. if you know a df doesn't have futures in it then quoting it should help a lot @Marvin can you show a concise example on that
m
✅ thought for 225.9 seconds
Totally—here’s a concise pattern for speeding things up with large objects by quoting them. In Prefect 3.x, 
quote
tells Prefect not to recursively inspect/resolve futures inside the value; the task still receives the raw value. Example with a large DataFrame:
Copy code
from prefect import flow, task
from prefect.utilities.annotations import quote
import pandas as pd

@task
def summarize(df):
    # Receives the raw DataFrame; quote is removed before invocation
    return df.shape

@flow
def main():
    df = pd.DataFrame({"a": range(5_000_000)})
    # If you know there are no futures in df, quote() avoids expensive introspection
    return summarize(quote(df))

if __name__ == "__main__":
    main()
Same idea for a big payload you don’t want Prefect to walk:
Copy code
@task
def consume(payload):
    # Receives the raw dict; no dependency tracking inside it
    return list(payload.keys())

@flow
def run():
    big_payload = {"df": pd.DataFrame({"a": range(1_000_000)}), "meta": {"x": 1}}
    return consume(quote(big_payload))
Notes: - Quoted values are not traversed for futures; dependencies inside are not tracked. - The task gets the unwrapped value (no 
quote
wrapper) at call time.
n
Awesome, thanks a bunch
n
catjam
n
Hi Nate, hope you had a great weekend. So I'm able to reproduce the wait (about 2 minutes) with code minimized to two files with a step that makes a dataframe that is 100 row * 46 column large. I've attempted to make a MRE from that using Co-pilot, but with the code that makes there is no wait. Can I send over the code that makes the 2 minute wait and the MRE and see if any of that helps identify the issue? I tried using quotes but what I'm passing between steps is a dataclass that has a data frame object within it, so quotes didn't seem to help.
Spoke too soon, able to reproduce a little in the MRE (About 1 min 30 seconds) I'll put that up soon
Ok, here's the code. WIth a data list of only 2 being passed it, it creates tasks quick. Get it up to 50 or 60 which is the usual amount this is mapped out to, and the wait to create tasks is about 1 minute 20 seconds. Not sure if that is to be expected or I should batch the list into groups of 10 or 20 as concurrency limits have the same issue.
copilot_mre_slow_mapping - Copy.py.txt
@Nate when you get a chance, can you please take a look at this. Today it took 20 minutes for prefect to start doing tasks after mapping started. I’m probably going to start getting in trouble with the investment research team soon. Seems like you were working on Sunday so not expecting any turn around tonight but it would really help to get some info on this tomorrow or Wednesday.
Not sure if screenshot helps:
n
if i copy and run your code exactly as written it executes in 13 seconds
Copy code
pbpaste | uv run --with pandas -
so this seems like its about your server installation, python environment, or something else not represented by your example
n
Gotcha, thanks for the quick turnaround, this is a tough thing to pin down
n
are you using tag based concurrency anywhere?
like when they are sitting and waiting, what state are the runs in?
n
The runs don’t seem to exist, the tag based stuff affects them when the start minutes later
My guess is maybe some id / fk generation issues in the sb since I see that error in the logs often
n
id / fk generation issues in the sb
what do you mean here?
/ what error are you seeing?
are you using postgres or sqlite for a db
n
In dev locally SQLite, everywhere else Postgres.
Seems to happen in both places, when I’m home I’ll post
Ok, ran the mre with a new environment with only prefect 3.6.4 and pandas installed, new server environment with a new sql lite db still taking about a minute to run. When waiting, it shows one task that's about to be mapped in the pending status. The Prefect server has (sqlite3.OperationalError) database is locked on a update deployment sql statement, which seems to happen quite often.
n
yes db locking on sqlite is a well-known issue. you should use postgres if you care about performance at non-demo scale
n
Yeah we do, weird it's happening on a database with zero deployments. Does telemetry run on prefect local server instances? We have some annoying cert permissions in place so maybe it's locking up when it tries to phone home?
Tried running it in debug mode, looks like the last time a task is submitted to the thread pool executor is 103214 AM and then the task run is created at 103347 AM. Is there any way to see what is happening between a task being submitted and a task run being created?
n
could it be thread pool exhaustion? what happens if you switch to a ProcessPoolTaskRunner (not that that's what you want, just wondering if somehow the thread pool has no threads to offer for a while, and so you're blocking on getting a thread for each task)
n
Thanks I'll give it a shot.
Ooof after 4 minutes only one task finished successfully.
n
hrm i suspect something is weird about the machine where you're running this code
or.... there's meaningful daylight between the MRE you shared and what you're running
n
Same MRE apart from the processpooltaskrunner
n
as in, the same one i ran? then i'd say its almost certainly about the machine where this is happening
or, as you mention, something about idiosyncratic networking? maybe? but idk why that'd cause a difference in behavior between threads/processes
n
I'll try it on the laptop I'm talking to you on, one sec
Ok same thing seems to be happening, but at least on this laptop I'm hitting prefect cloud so hopefully you can see more
Still a minute 30 seconds for things to start up
Here's the code again if that helps. Swear I didn't change anything.
copilot_mre_slow_mapping.py
n
Copy code
» pbpaste | uv run --with pandas -
sorry, im not sure what could be happening in your case
n
Thanks for looking at it. Anyone I could raise this to? New virtual environment with only pandas and prefect 3.6.4 installed, seem to be able to reproduce it over multiple machines, even if it hits prefect cloud? Hard to know where to look to try to fix the issue.
Just weird the same code can run so differently:
I mean could it be the package manager itself?
n
by package manager, you mean 
pip
? or
if the hang is between submission and the task running, that'd be surprising to me
n
Yeah pip vs uv, unsure if that would make a difference if virtual environments were different. Yeah me too, but willing to try anything lol
n
odd. lemme share internally to see if anyone has ideas
question: do you have docker installed on your machine?
n
Not on this one
n
doh, ok i wanted you to try this
Copy code
pbpaste | docker run -i --rm prefecthq/prefect:3-python3.12 uv run --with pandas -
after copying your MRE to the clipboard
n
Gotcha I can get it on my home laptop, gimme some time to get that set up. Docker Desktop ok?
n
yep yep
n
Ok probably have some news in the afternoon
n
🫡
n
On temporary server it seems to run fine:
Going to try running it against the prefect cloud server
Using the prefect docker image is definitely far faster than using a virtual environment:
My virtual environment is on 3.13.1, if anything I'd be expecting it to be faster:
Got a smaller MRE now, runs on docker with uv connected to prefect cloud in under 2 seconds. Runs in 50 seconds with a new virtual environment hitting prefect cloud:
Copy code
import time
from prefect import flow, task

@task
def dummy_task(x):
    return x

@flow
def test_mapping():
    start = time.time()
    futures = dummy_task.map(range(100))
    map_time = time.time() - start
    print(f"Mapping took: {map_time:.2f}s")
    
    start = time.time()
    futures.wait()
    wait_time = time.time() - start
    print(f"Waiting took: {wait_time:.2f}s")

if __name__ == "__main__":
    test_mapping()
n
is this..... a windows thing?
😕
n
Claude is saying I should downgrade from 3.13.1 to 3.12.12 like your docker env
Going to give that a shot, hopefully it's not a windows thing
n
i would be greatly surprised if that made a difference, though you can trivially try via 
--python 3.x.y
with uv run
n
I got nothing, running example code with this:
Copy code
Get-Clipboard | docker run -i -e PREFECT_API_URL=$env:PREFECT_API_URL -e PREFECT_API_KEY=$env:PREFECT_API_KEY --rm prefecthq/prefect:3-python3.12 uv run --with pandas --with prefect --python 3.13.1 -
Gets me: Mapping took: 0.22s Waiting took: 4.01s
@Nate Are you able to see Prefect cloud logs? Running through windows on clean virtual env I have this flow run: https://app.prefect.cloud/account/d5601f6d-e676-4fdb-874d-c01da8f9d469/workspace/ec4a[…]runs/flow-run/06926127-db79-7f8a-8000-14fe40d25325?tab=split -- Ran it in debug mode so there's lots more logs. But the issue is definitely on the client side it seems, getting lots of attempts to connect to the api, seems like it connects for every task that is to be mapped:
n
sorry, i am not, but i agree it seems like a client-side thing so im not sure what the cloud backend logs would offer in this case .... so far it does seem like something that's specific to your machine. later i can try running your MRE in github actions on a windows machine to see if (🤞 hopefully not) its a windows thing
n
Connections to the API get created soooooo much faster through the Docker image. Updated code to log times at the end.
Copy code
import os
# MUST be before any prefect imports

import time
from prefect import flow, task, get_run_logger


@task
def dummy_task(x):
    return x

@flow
def test_mapping():
    logger = get_run_logger()
    start = time.time()
    futures = dummy_task.map(range(100))
    map_time = time.time() - start
    <http://logger.info|logger.info>(f"Mapping took: {map_time:.2f}s")
    
    start = time.time()
    futures.wait()
    wait_time = time.time() - start
    <http://logger.info|logger.info>(f"Waiting took: {wait_time:.2f}s")

    <http://logger.info|logger.info>(f"Final Notes:")
    <http://logger.info|logger.info>(f"Mapping took: {map_time:.2f}s")
    <http://logger.info|logger.info>(f"Waiting took: {wait_time:.2f}s")
if __name__ == "__main__":
    test_mapping()
How I'm running it through docker to get to the cloud:
Copy code
Get-Clipboard | docker run -i -e PREFECT_API_URL=$env:PREFECT_API_URL -e PREFECT_API_KEY=$env:PREFECT_API_KEY -e PREFECT_LOGGING_LEVEL="DEBUG" --rm prefecthq/prefect:3-python3.12 uv run --with pandas --with prefect --python 3.13.1 -
At this point all I'm doing is trying to figure out with claude why the network traffic is different between windows and docker, something about the way tcp is handled apparently.
Linux/docker connections are made in milliseconds, windows ones are like 0.7 seconds each:
Ok, got a github action that runs both linux and windows. Windows is noticably slower: got 37.9Seconds on the wait. It's a public github so hopefully you can see it: https://github.com/Octacon100/mre_slow_windows_mapping_prefect/actions/runs/19706302931/job/56454673317 Windows: (See image)
Linux -
@Nate, hope you had a good Thanksgiving weekend, should I make a git hub issue about this or something? I'm not really sure what next steps with this are if I can reproduce it on github actions.
n
yes, a github issue would make sense! thank you
n
Ok, made an issue, hopefully that helps: https://github.com/PrefectHQ/prefect/issues/19581
3 Views
Open in Slack
PreviousNext
Powered by