https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> 00:05:58.209 | INFO | Task run 'pr...
# ask-marvin
o
@Marvin 000558.209 | INFO | Task run 'produce_trades-389' - Finished in state Completed() 000947.164 | ERROR | Flow run 'brilliant-rottweiler' - Encountered exception during execution: PrefectHTTPStatusError("Server error '500 Internal Server Error' for url 'http://host.docker.internal:4200/api/flow_runs/'\nResponse: {'exception_message': 'Internal Server Error'}\nFor more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/500") Traceback (most recent call last): File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 781, in run_context yield self File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 1395, in run_flow_sync engine.call_flow_fn() File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 801, in call_flow_fn result = call_with_parameters(self.flow.fn, self.parameters) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/utilities/callables.py", line 210, in call_with_parameters return fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/app/qfin-master/brrr/orchestration/prefect_enhanced_flow.py", line 32, in contextvar_wrapper return original_fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/app/qfin-master/brrr/data_collection/nordnet/prefect_deploys/nn_get_nordic_trades_prefect.py", line 176, in nn_get_nordic_trades_prefect transform_and_write_to_database(raw_payloads, db_type, logger) File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/flows.py", line 1702, in call return run_flow( ^^^^^^^^^ File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 1552, in run_flow ret_val = run_flow_sync(**kwargs) ^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 1392, in run_flow_sync with engine.start(): ^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/contextlib.py", line 137, in enter return next(self.gen) ^^^^^^^^^^^^^^ File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 758, in start with self.initialize_run(): ^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/contextlib.py", line 137, in enter return next(self.gen) ^^^^^^^^^^^^^^ File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 691, in initialize_run self.flow_run = self.create_flow_run(self.client) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 528, in create_flow_run return client.create_flow_run( ^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/client/orchestration/_flow_runs/client.py", line 119, in create_flow_run response = self.request("POST", "/flow_runs/", json=flow_run_create_json) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/client/orchestration/base.py", line 33, in request return self._client.send(request) ^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/client/base.py", line 622, in send response.raise_for_status() File "/opt/prefect/app/.venv/lib/python3.12/site-packages/prefect/client/base.py", line 163, in raise_for_status raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.cause prefect.exceptions.PrefectHTTPStatusError: Server error '500 Internal Server Error' for url 'http://host.docker.internal:4200/api/flow_runs/' Response: {'exception_message': 'Internal Server Error'} For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/500 000951.825 | INFO | Flow run 'brilliant-rottweiler' - Finished in state Failed("Flow run encountered an exception: PrefectHTTPStatusError: Server error '500 Internal Server Error' for url 'http://host.docker.internal:4200/api/flow_runs/'\nResponse: {'exception_message': 'Internal Server Error'}\nFor more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/500") 000951.828 | ERROR | Flow run 'brilliant-rottweiler' - Engine execution exited with unexpected exception Traceback (most recent call last):
m
✅ thought for 360.7 seconds
Thanks for sharing the trace — this error is happening when your flow tries to create a subflow run (POST /api/flow_runs/) for 
transform_and_write_to_database
. In Prefect 3.x, the most common cause of a 500 at this exact point is sending very large parameters to the subflow. Why this often happens - Subflows serialize their parameters and send them to the API. If the serialized JSON is too large (e.g., big lists, dataframes, or nested dicts), the server can error. In Prefect 3.x, flow run parameters have a hard cap (on the order of hundreds of KB). Passing large payloads (like the full 
raw_payloads
) to another flow will frequently trigger this. Docs: Flows and subflows (Prefect 3.x) How to confirm quickly 1) Check the size of what you’re passing to the subflow:
Copy code
import json
def json_size_kb(obj) -> float:
    return len(json.dumps(obj, default=str).encode("utf-8")) / 1024

print("param size KB:", json_size_kb({"raw_payloads": raw_payloads}))
If it’s hundreds of KB or more, that’s very likely the culprit. 2) Check server logs for the actual exception - If you’re running a local server on the host, run it with debug logging and watch the console:
Copy code
prefect server start --log-level DEBUG
- If you’re running the server in Docker, view logs from that container:
Copy code
docker logs -f <prefect-server-container-name>
You should see the traceback behind the 500 response there. 3) Verify client/server versions align (both Prefect 3.x)
Copy code
# client
python -c "import prefect; print(prefect.__version__)"

# server
curl <http://host.docker.internal:4200/api/version>
Significant mismatches can cause unexpected server errors. Recommended patterns to fix/avoid this - Prefer tasks over subflows for heavy data. Tasks run in-process and don’t serialize their inputs to the API:
Copy code
from prefect import flow, task

@task
def transform_and_write_to_database(raw_payloads, db_type):
    # do the heavy work here
    ...

@flow
def nn_get_nordic_trades_prefect(...):
    raw_payloads = ...
    transform_and_write_to_database(raw_payloads, db_type)
- Pass lightweight references instead of big objects. For example, persist the large payload to storage, then pass the path or key to the subflow:
Copy code
from prefect import flow, task

@task
def stage_payloads(raw_payloads) -> str:
    # write to S3/GCS/local path and return the reference
    path = "<s3://my-bucket/run-123/payloads.json>"
    ...
    return path

@flow
def nn_get_nordic_trades_prefect(...):
    raw_payloads = ...
    ref = stage_payloads(raw_payloads)
    transform_and_write_to_database_flow(data_ref=ref, db_type=db_type)

@flow
def transform_and_write_to_database_flow(data_ref: str, db_type: str):
    # read from data_ref inside the subflow
    ...
- If you really want subflow observability, still keep the subflow params small and load the heavy data inside the subflow from a reference (file path, S3 key, table name, etc.). A couple of quick tests you can try now - Swap 
@flow
to 
@task
on 
transform_and_write_to_database
and call it from the parent flow. If the 500 disappears, it confirms the large-parameter subflow path was the problem. - Keep 
@flow
but change the call to pass only a small reference (e.g., 
data_ref
) and load from that reference inside the subflow. Helpful docs - Flows and subflows: https://docs.prefect.io/v3/concepts/flows - Tasks: https://docs.prefect.io/v3/concepts/tasks Questions to help me narrow this down for you - What is the approximate size/type of 
raw_payloads
you’re passing to 
transform_and_write_to_database
? - Are your client and server both Prefect 3.x? What versions are they on? - Are you running the server outside the container (accessed via host.docker.internal) and can you share any relevant stack traces from the server logs around the 500?
If you share a snippet of the parent flow call site and the 
transform_and_write_to_database
signature, I can suggest the smallest change to avoid the large payload while keeping your structure.
o
@Marvin I see. So just using sub flows to merely organize tasks would be a bad design if it involved passing large payloads, in general?
m
✅ thought for 19.8 seconds
Short answer: yes—if the only reason for a subflow is “grouping” and it forces you to pass large payloads between flows, that’s a poor fit. Guidance: - Use tasks to organize work inside a single flow run. Task inputs stay in-process, so you avoid API serialization limits and overhead. - Use subflows when you need their specific benefits: separate observability/run records, different retries/schedules/tags, isolation via different task runners/infrastructure, or reuse as a callable flow. If you do, keep parameters small. - To keep subflows AND handle big data, pass lightweight references instead of the data itself: - Persist the data (e.g., S3/GCS/filepath/table name) and pass the path/key/ID. - Or rely on task result storage so the subflow receives a reference and loads the data when it runs. - If you just need parallelism/chunking, prefer mapped tasks or chunked task calls over subflows. Helpful docs: - Flows and subflows: Flows (Prefect 3.x) - Tasks: Tasks (Prefect 3.x) If you share roughly how large 
raw_payloads
is and what you want from the subflow (e.g., separate retries vs. just grouping), I can suggest the smallest change: switch to a task, or keep the subflow and pass a reference.
o
@Marvin can you give me some guidance on how i can use task result storage and passing a reference as well as "mapped tasks" (whatever that is)
m
✅ thought for 147.3 seconds
Absolutely—here’s a quick, practical guide for each: 1) Task result storage (store big results, pass small references) - Goal: avoid sending large data as flow/subflow parameters by letting Prefect persist task results to storage and pass references (futures) between tasks/flows. - How it works in 3.x: - Tasks can persist their return values via a result store (local, S3, GCS, etc.). - Downstream tasks/subflows receive a small reference; the engine loads the actual value at execution time. - Minimal example:
Copy code
from prefect import flow, task
from prefect.results import get_default_result_storage  # optional

# Optional: set a specific storage for this task's results
result_storage = get_default_result_storage()  # or your custom storage

@task(persist_result=True, result_storage=result_storage)  # persist this task's result
def produce_large_data() -> dict:
    # Imagine this returns a big payload
    return {"rows": [...]}  # large

@flow
def parent_flow():
    # Returns a PrefectFuture; reference is small
    large_data_ref = produce_large_data()
    # Pass the reference to a subflow or task; value is loaded when used
    return consume_large_data(large_data_ref)

@task
def consume_large_data(large_data: dict) -> int:
    # Prefect resolves the reference and reads from storage
    return len(large_data["rows"])
- Notes: - 
persist_result=True
is the key; Prefect chooses a serializer and storage (configurable). - For custom backends, set 
result_storage
and 
result_serializer
. - This is the preferred pattern when you still want to use subflows but avoid big params. 2) Passing a lightweight reference (DIY path/key instead of data) - Goal: you control where the data lives (S3, GCS, DB, local file) and pass a small pointer. - Example:
Copy code
from prefect import flow, task, unmapped

@task
def stage_payload(raw_payloads) -> str:
    # Write payload once and return a path/key
    path = "/tmp/payload.json"   # or <s3://bucket/key>
    with open(path, "w") as f:
        f.write(serialize(raw_payloads))
    return path

@task
def transform_and_write(path: str, db_type: str):
    raw = deserialize(open(path).read())
    # do transform + write to db

@flow
def nn_get_nordic_trades_prefect(db_type: str):
    raw_payloads = ...  # large
    ref = stage_payload(raw_payloads)     # small reference
    transform_and_write(ref, db_type)     # safe to pass around
- Variation: If you keep the subflow, pass 
path
to the subflow instead of the payload. 3) Mapped tasks (parallelize work over iterables) - What “mapped tasks” means: run the same task many times concurrently over an iterable input using 
.map()
. - When to use: fan-out over items/chunks; works great with ThreadPoolTaskRunner (I/O-bound) or ProcessPoolTaskRunner (CPU-bound). - Basic mapping:
Copy code
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def fetch_one(symbol: str) -> dict:
    # call API, return record
    ...

@flow(task_runner=ThreadPoolTaskRunner(max_workers=10))
def pull_trades(symbols: list[str]) -> list[dict]:
    futures = fetch_one.map(symbols)   # runs concurrently
    return futures.result()            # blocks for all
- Mapping with a large constant using unmapped():
Copy code
from prefect import flow, task, unmapped

@task
def transform_chunk(chunk_path: str, config: dict) -> int:
    data = read_chunk(chunk_path)
    return apply_transform(data, config)

@flow
def process_all(chunks: list[str], big_config: dict):
    futures = transform_chunk.map(
        chunk_path=chunks,
        config=unmapped(big_config)  # don’t replicate big_config N times
    )
    return futures.result()
- Choosing task runners: - ThreadPoolTaskRunner: best for I/O-bound (web calls, DB I/O). - ProcessPoolTaskRunner: best for CPU-bound; params/results must be picklable. Putting it together for your case - If your “subflow” was just grouping tasks, convert it to tasks and, if needed, map over chunks. - To keep subflows:
- Persist large results in upstream tasks (
persist_result=True
) and pass the futures. - Or stage the payload (file/S3/DB) and pass only a path/key. Common pitfalls to avoid - Passing big dicts/lists directly to subflows (will hit parameter limits). - Mapping a huge constant accidentally; wrap constants with 
unmapped()
. - Using ProcessPool for non-picklable objects. If you share: - Where you’d like the result storage (local path, S3, etc.) - Whether your transforms are I/O-bound or CPU-bound - How you want to chunk the data …I’ll sketch an example tailored to your repository and suggest exact settings for 
persist_result
, storage, and task runner.
3 Views
Open in Slack
PreviousNext
Powered by