https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> in the @ ecs decorator, how would I ...
# ask-marvin
m
@Marvin in the @ ecs decorator, how would I set the job variables for network config and subnets and security groups?
m
✅ thought for 67.0 seconds
With the 
@ecs
decorator, you pass them as job variables directly on the decorator. Example:
Copy code
python
from prefect import flow
from prefect_aws.decorators import ecs

@ecs(
    work_pool="my-ecs-pool",
    vpc_id="vpc-12345678",
    network_configuration={
        "subnets": [
            "subnet-12345678",
            "subnet-87654321",
        ],
        "securityGroups": [
            "sg-12345678",
            "sg-87654321",
        ],
        "assignPublicIp": "ENABLED",
    },
)
@flow
def my_flow():
    pass
A few important notes: - 
network_configuration
is the key you want - inside it, use: - 
subnets
- 
securityGroups
- 
assignPublicIp
- if you're setting custom networking, include 
vpc_id
as well So the relevant part is:
Copy code
python
@ecs(
    work_pool="my-ecs-pool",
    vpc_id="vpc-12345678",
    network_configuration={
        "subnets": ["subnet-1", "subnet-2"],
        "securityGroups": ["sg-1", "sg-2"],
        "assignPublicIp": "DISABLED",
    },
)
If helpful, I can also show you the equivalent setup in: - 
prefect.yaml
- 
flow.deploy(...)
- an ECS work pool's base job template Docs: - Prefect AWS ECS worker docs - Prefect docs
m
Ok, @Marvin I have this error with the decorator 20 May 2026, 10:25 092555.250 | ERROR | prefect.engine - Validation of flow parameters failed with error: cannot specify both default and default_factory prefect 20 May 2026, 10:25 <frozen runpy>128 RuntimeWarning: 'prefect_aws.experimental.bundles.execute' found in sys.modules after import of package 'prefect_aws.experimental.bundles', but prior to execution of 'prefect_aws.experimental.bundles.execute'; this may result in unpredictable behaviour
m
✅ thought for 146.1 seconds
That error is probably not coming from your flow code itself — it looks like a schema/validation issue triggered by the ECS decorator's job variables, especially 
network_configuration
. What’s likely happening In 
prefect-aws
, some ECS job variable fields are defined with 
default_factory
(including 
network_configuration
). During parameter/schema generation, Prefect ends up with a field that effectively has both a 
default
and a 
default_factory
, and Pydantic rejects that with:
Copy code
text
cannot specify both default and default_factory
So this is likely an issue in the interaction between the 
@ecs
decorator / deployment parameter handling and the ECS worker variable schema — not that your subnet/security group values are malformed. Most likely trigger Using something like this on the decorator:
Copy code
python
@ecs(
    work_pool="my-ecs-pool",
    vpc_id="vpc-12345678",
    network_configuration={
        "subnets": [...],
        "securityGroups": [...],
        "assignPublicIp": "ENABLED",
    },
)
may be enough to trigger it. Things to try right now 1. Use 
vpc_id
only first Remove 
network_configuration
temporarily and see if the deployment/run succeeds:
Copy code
python
from prefect import flow
from prefect_aws.decorators import ecs

@ecs(
    work_pool="my-ecs-pool",
    vpc_id="vpc-12345678",
)
@flow
def my_flow():
    pass
If that works, it strongly points to 
network_configuration
as the culprit. 2. Move networking config out of the decorator and into the work pool / deployment job variables Instead of putting it on the decorator, set it in 
prefect.yaml
or the work pool configuration. Example in `prefect.yaml`:
Copy code
yaml
deployments:
  - name: my-deployment
    entrypoint: flows.py:my_flow
    work_pool:
      name: my-ecs-pool
      job_variables:
        vpc_id: vpc-12345678
        network_configuration:
          subnets:
            - subnet-12345678
            - subnet-87654321
          securityGroups:
            - sg-12345678
          assignPublicIp: ENABLED
3. Make sure 
network_configuration
is not also a flow parameter For example, avoid:
Copy code
python
@flow
def my_flow(network_configuration: dict = {}):
    pass
Use:
Copy code
python
@flow
def my_flow():
    pass
About the RuntimeWarning
Copy code
text
RuntimeWarning: 'prefect_aws.experimental.bundles.execute' found in sys.modules...
That looks secondary. It suggests some experimental bundles module is being imported in a way Python doesn't like, but it's probably not the root cause of the validation failure. What I’d recommend If you want the quickest path forward: - remove 
network_configuration
from the 
@ecs
decorator - keep 
vpc_id
there if needed - set subnets / security groups in the deployment or work pool job variables instead If you want, paste your actual 
@ecs(...)
decorator block and I can help rewrite it into a form that should avoid this error. Useful links: - Prefect AWS ECS worker docs - Prefect docs
m
@Marvin ok great guess, I was doing that, I have moved them out and only left cpu/memory. I am still getting the same error? @ecs( work_pool=f"wp_name", cpu=16384, memory=65536, environment={ "PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_SQS_QUEUE_NAME": f"sqs_name" } )
m
✅ thought for 143.8 seconds
Yep — I see one likely problem immediately: for the ECS decorator, the field is 
env
, not 
environment
. So this:
Copy code
python
@ecs(
    work_pool="wp_name",
    cpu=16384,
    memory=65536,
    environment={
        "PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_SQS_QUEUE_NAME": "sqs_name"
    }
)
should be:
Copy code
python
from prefect import flow
from prefect_aws.decorators import ecs

@ecs(
    work_pool="wp_name",
    cpu=16384,
    memory=65536,
    env={
        "PREFECT_INTEGRATIONS_AWS_ECS_OBSERVER_SQS_QUEUE_NAME": "sqs_name"
    },
)
@flow
def my_flow():
    pass
Why this matters The ECS worker/job variable model uses 
env
, and that field already has a 
default_factory=dict
. Passing an unexpected field like 
environment
can contribute to schema generation issues and may be related to the validation error you're seeing. Next thing to check If changing 
environment
-> 
env
still gives the same error, then the problem may be a broader bug in the current 
prefect-aws
/ Prefect version combination, not your specific values. In that case, I’d test these in order: 1. Remove 
env
too, leaving only cpu/memory
Copy code
python
@ecs(
    work_pool="wp_name",
    cpu=16384,
    memory=65536,
)
2. If that still fails, remove cpu/memory as well and test the bare minimum:
Copy code
python
@ecs(work_pool="wp_name")
That will tell us whether: - 
env
was the issue - one of 
cpu
/ 
memory
is also triggering the same schema bug - or the decorator itself is broken in your installed version My best guess right now There may be a version-specific bug where the decorator’s variable schema conflicts with Pydantic, even for fields like: - 
env
- 
cpu
- 
memory
So the most useful next step is: 1. change 
environment
to 
env
2. if still broken, try 
@ecs(work_pool="wp_name")
3. tell me: - your 
prefect
version - your 
prefect-aws
version You can get them with:
Copy code
bash
pip show prefect prefect-aws
or
Copy code
bash
uv pip show prefect prefect-aws
If you paste those versions, I can help narrow down whether this is a known compatibility issue. Reference: - Prefect AWS ECS worker docs - Prefect docs
m
@Marvin I removed them and made sure they were set as defaults in the template and no change
Copy code
Name: prefect
Version: 3.7.1
Summary: Workflow orchestration and management.
Home-page:
Author:
Author-email: "Prefect Technologies, Inc." <help@prefect.io>
License: Apache-2.0
Location: /<redacted>/.venv/lib/python3.13/site-packages
Requires: aiosqlite, alembic, amplitude-analytics, anyio, apprise, asgi-lifespan, asyncpg, cachetools, click, cloudpickle, coolname, cryptography, cyclopts, dateparser, docker, exceptiongroup, fastapi, fsspec, graphviz, griffe, httpcore, httpx, humanize, jinja2, jinja2-humanize-extension, jsonpatch, jsonschema, opentelemetry-api, orjson, packaging, pathspec, pluggy, prometheus-client, pydantic, pydantic-core, pydantic-extra-types, pydantic-settings, pydocket, python-dateutil, python-slugify, pytz, pyyaml, readchar, rfc3339-validator, rich, ruamel-yaml, ruamel-yaml-clib, semver, sniffio, sqlalchemy, toml, typing-extensions, uvicorn, websockets, whenever
Required-by: prefect-aws, prefect-docker, prefect-slack
---
Name: prefect-aws
Version: 0.7.9
Summary: Prefect integrations for interacting with Amazon Web Services.
Home-page: <https://github.com/PrefectHQ/prefect/tree/main/src/integrations/prefect-aws>
Author:
Author-email: "Prefect Technologies, Inc." <help@prefect.io>
License: Apache License 2.0
Location: /<redacted>/Projects/mwi-data-pipelines/.venv/lib/python3.13/site-packages
Requires: aiobotocore, boto3, botocore, prefect, pyparsing, rich, tenacity, typer
Required-by:
m
🔄 Researching your answer... this may take a while 🔧 Using tool: 
read_github_issues
📊 Tools used: • 
read_github_issues
(2x)
✅ thought for 251.9 seconds
Thanks — that version info is very helpful. This now looks much more like a Prefect 3.7.1 + Python 3.13 validation bug than an ECS configuration problem. What’s going on The strongest signal here is: - you removed 
network_configuration
- then removed 
environment
/ 
env
- then moved defaults into the template - and the error did not change That suggests the failure is not caused by ECS job variables anymore. Based on the code path in Prefect 3.7.1, Python 3.13 can hit a buggy parameter-validation path where Prefect mishandles Pydantic 
Field(...)
metadata on flow parameters and ends up constructing a field with both: - 
default
- 
default_factory
which produces exactly:
Copy code
text
cannot specify both default and default_factory
Most likely culprit now Look at your flow function signature, not the 
@ecs
decorator. Specifically, check whether any flow parameters are declared with Pydantic 
Field(...)
, for example:
Copy code
python
from pydantic import Field
from prefect import flow

@flow
def my_flow(
    x: str = Field(default="abc", description="example"),
):
    ...
or anything similar using: - 
Field(default=...)
- 
Field(default_factory=...)
- 
Annotated[..., Field(...)]
Under your current stack: - 
prefect==3.7.1
- 
prefect-aws==0.7.9
- Python 3.13 that appears to be the more likely trigger. What to try next 1. Temporarily remove all 
Field(...)
usage from flow parameters Change this kind of pattern:
Copy code
python
from pydantic import Field
from prefect import flow

@flow
def my_flow(
    name: str = Field(default="world", description="Name"),
):
    ...
to plain Python defaults:
Copy code
python
from prefect import flow

@flow
def my_flow(
    name: str = "world",
):
    ...
If you need descriptions, put them in the docstring instead. 2. Test the flow with the bare minimum ECS decorator
Copy code
python
from prefect import flow
from prefect_aws.decorators import ecs

@ecs(work_pool="wp_name")
@flow
def my_flow(name: str = "world"):
    return name
3. If that works, add back flow params gradually, but without 
Field(...)
. Why I think this is the issue Because if the decorator args were the problem, removing them should have changed behavior. Since it didn’t, the next most likely source is flow parameter schema generation. And your environment is exactly the kind that can expose this: - newer Prefect - newer Pydantic behavior - Python 3.13 Workarounds In order of practicality: 1. Best immediate workaround: avoid 
pydantic.Field(...)
in flow parameters 2. If possible: run this flow under Python 3.12 instead of 3.13 3. Alternative: deploy/run without the 
@ecs
decorator temporarily to confirm the flow itself validates cleanly What I’d like to see next If you paste just the flow function signature (you can redact names/business logic), I can tell you exactly which parameter definition is likely triggering this. For example, paste something like:
Copy code
python
@ecs(work_pool="...")
@flow
def my_flow(
    ...
):
    ...
and I’ll help pinpoint the bad field quickly. Relevant docs: - Prefect docs - Prefect AWS ECS worker docs
m
@Marvin
Copy code
@ecs(
        work_pool=f"mwi-data-pipelines-{env_switch}-primary",
)
@flow(flow_run_name="global_pipeline_{storage_key}_{partition_date}", name="global_pipeline")
def global_pipeline(global_resources:GlobalResources, tasks: list[str], storage_key: str, partition_date: str, other_args: Optional[dict[str, Any]] = None):
m
✅ thought for 12.3 seconds
I think I see it. Your 
flow_run_name
is written as a plain string:
Copy code
python
@flow(flow_run_name="global_pipeline_{storage_key}_{partition_date}", name="global_pipeline")
In Prefect, 
flow_run_name
should usually be either: - a format string that Prefect can resolve from parameters, or - a callable But here the bigger issue is likely your parameter types, especially this one:
Copy code
python
global_resources: GlobalResources
and possibly:
Copy code
python
other_args: Optional[dict[str, Any]] = None
The 
@ecs
wrapper causes Prefect to build a parameter schema for the flow, and if one of those parameter types is a Pydantic model / dataclass / custom type with 
Field(...)
usage or default factories inside it, Prefect 3.7.1 on Python 3.13 can blow up with:
Copy code
text
cannot specify both default and default_factory
So the likely problem is inside 
GlobalResources
, not the decorator. What to inspect in 
GlobalResources
 Please check whether that model/class contains fields like:
Copy code
python
from pydantic import BaseModel, Field

class GlobalResources(BaseModel):
    x: dict[str, str] = Field(default_factory=dict)
    y: str = Field(default="abc")
or any combination of: - 
Field(default_factory=...)
- 
Field(default=...)
- inherited Pydantic fields - 
Annotated[..., Field(...)]
That’s the strongest suspect. Quick way to confirm Try temporarily replacing the signature with only plain built-in types:
Copy code
python
from typing import Any, Optional
from prefect import flow
from prefect_aws.decorators import ecs

@ecs(work_pool=f"mwi-data-pipelines-{env_switch}-primary")
@flow(flow_run_name="global_pipeline_{storage_key}_{partition_date}", name="global_pipeline")
def global_pipeline(
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    pass
If that works, then 
GlobalResources
is the trigger. If it still fails, next reduce further to:
Copy code
python
@ecs(work_pool=f"mwi-data-pipelines-{env_switch}-primary")
@flow(name="global_pipeline")
def global_pipeline(
    storage_key: str,
    partition_date: str,
):
    pass
That will isolate whether: - 
GlobalResources
is the issue - 
flow_run_name
templating is involved - or there is a deeper validation bug Most likely root cause ranking 1. 
GlobalResources
contains Pydantic 
Field(...)
/ 
default_factory
usage 2. Prefect 3.7.1 + Python 3.13 parameter validation bug 3. less likely: 
other_args
or decorator config What I recommend next Please paste the definition of 
GlobalResources
(redacted if needed). That is the most likely place the bad field is hiding. If you want, I can also help you rewrite 
GlobalResources
into a shape that avoids this Prefect/Pydantic edge case. Docs: - Prefect docs - Prefect AWS ECS worker docs
m
GlobalResources
is a dataclass @Marvin
m
✅ thought for 10.0 seconds
That still fits the likely root cause. A dataclass can absolutely trigger this too — especially if it uses 
field(default_factory=...)
. So the next thing I’d inspect is whether 
GlobalResources
looks anything like:
Copy code
python
from dataclasses import dataclass, field

@dataclass
class GlobalResources:
    items: dict[str, str] = field(default_factory=dict)
or has any: - 
field(default_factory=...)
- inherited dataclass fields - nested dataclasses / Pydantic models - mutable defaults being handled via factories Given your error:
Copy code
text
cannot specify both default and default_factory
and the fact that stripping ECS job variables didn’t help, the most likely explanation is: Prefect 3.7.1 on Python 3.13 is choking while building the flow parameter schema for 
GlobalResources
. Fastest way to prove it Temporarily change the parameter to something simpler. Instead of:
Copy code
python
def global_pipeline(
    global_resources: GlobalResources,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
try:
Copy code
python
def global_pipeline(
    global_resources: dict[str, Any],
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
If that works, 
GlobalResources
is confirmed as the trigger. What to do if 
GlobalResources
is the issue You’ve got a few options: 1. Pass a plain dict instead of the dataclass Simplest workaround. 2. Serialize before passing Convert the dataclass to a dict before calling the flow:
Copy code
python
   from dataclasses import asdict

   global_pipeline(
       global_resources=asdict(resources),
       ...
   )
3. Use Python 3.12 instead of 3.13 If changing the type is inconvenient, this may avoid the buggy validation path. 4. Avoid dataclass fields with 
default_factory
 If feasible, simplify the dataclass. Recommended immediate test Try this exact change first:
Copy code
python
from typing import Any, Optional
from prefect import flow
from prefect_aws.decorators import ecs

@ecs(work_pool=f"mwi-data-pipelines-{env_switch}-primary")
@flow(flow_run_name="global_pipeline_{storage_key}_{partition_date}", name="global_pipeline")
def global_pipeline(
    global_resources: dict[str, Any],
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    pass
If that succeeds, we’ve isolated it. If you paste the 
GlobalResources
dataclass definition, I can tell you exactly which field is the likely offender and suggest the least invasive fix.
m
@dataclass() class GlobalResources: athena_client: AthenaResource mongo_data: MongoResource s3_client: Boto3Resource each sub class is a dataclass also but I don't use pydantic on this set up just the python dataclass @Marvin
m
✅ thought for 15.3 seconds
Yep — that’s very likely the problem. 
GlobalResources
is not just a simple data object; it contains live resource objects: - 
AthenaResource
- 
MongoResource
- 
Boto3Resource
Even though they’re plain dataclasses, Prefect still has to build a parameter schema for the flow, and those nested dataclass/resource types are not good flow parameters for a deployed/ECS flow. Two important reasons: 1. Flow parameters need to be schema-validated / serialized For a deployment, Prefect treats flow arguments as inputs that may be serialized and sent to remote infrastructure. Resource/client objects usually don’t fit that model. 2. These are runtime resources, not parameters Things like boto3 clients, DB handles, service wrappers, etc. should generally be created inside the flow/task, not passed as flow parameters. So this is almost certainly why the 
@ecs
-wrapped flow is failing validation. What to do instead Pass only serializable config into the flow, then construct the resources inside the flow. For example, instead of this:
Copy code
python
@dataclass()
class GlobalResources:
    athena_client: AthenaResource
    mongo_data: MongoResource
    s3_client: Boto3Resource

@ecs(work_pool="...")
@flow(...)
def global_pipeline(
    global_resources: GlobalResources,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    ...
do something like this:
Copy code
python
from dataclasses import dataclass
from typing import Any, Optional
from prefect import flow
from prefect_aws.decorators import ecs

@dataclass
class GlobalResourceConfig:
    aws_region: str
    mongo_db_name: str
    bucket_name: str

@ecs(work_pool="my-pool")
@flow(flow_run_name="global_pipeline_{storage_key}_{partition_date}", name="global_pipeline")
def global_pipeline(
    resource_config: dict[str, Any],
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    athena_client = make_athena_resource(resource_config)
    mongo_data = make_mongo_resource(resource_config)
    s3_client = make_s3_resource(resource_config)

    ...
Or even simpler, don’t pass resource config at all if it can be read from env / blocks / secrets inside the flow. Best practice for Prefect deployments / remote execution For remotely executed flows, parameters should usually be: - strings - numbers - booleans - lists / dicts of simple values - small serializable config objects Avoid passing: - boto3 clients - database connections - SDK resource wrappers - service objects - instantiated classes representing live runtime state Practical recommendation for your case I would change this:
Copy code
python
global_resources: GlobalResources
to either:
Copy code
python
global_resources: dict[str, Any]
or better, remove it entirely and rebuild the resources inside the flow. For example:
Copy code
python
from typing import Any, Optional
from prefect import flow
from prefect_aws.decorators import ecs

@ecs(work_pool=f"mwi-data-pipelines-{env_switch}-primary")
@flow(flow_run_name="global_pipeline_{storage_key}_{partition_date}", name="global_pipeline")
def global_pipeline(
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    athena_client = build_athena_resource()
    mongo_data = build_mongo_resource()
    s3_client = build_s3_resource()

    ...
Why it may have seemed unrelated to ECS before Without remote infrastructure, passing rich Python objects around can sometimes appear to work locally. But once Prefect needs to validate/deploy/run the flow in ECS, those parameters have to go through a stricter schema/serialization path. That’s where this starts breaking. My confidence level High that 
global_resources
is the real issue now. If you want, I can help you refactor this flow signature into a Prefect-friendly pattern using: - env vars - Prefect blocks
- lazy resource initialization inside the flow - a lightweight serializable config dataclass/dict
m
ok, I moved that down into the flow. It's no longer launching the ecs tasks but is now failing to pickle @Marvin def global_pipeline(prefect_env: str, tasks: list[str], storage_key: str, partition_date: str, other_args: Optional[dict[str, Any]] = None):
Copy code
futures.append(global_pipeline.submit_to_work_pool(
            prefect_env=prefect_env,
            tasks=tasks if tasks else ['global_tasks'],
            storage_key=storage_key,
            partition_date=partition_date,
            other_args=other_args,
        ))
Engine execution exited with unexpected exception Traceback (most recent call last): File "/usr/local/lib/python3.12/site-packages/prefect/flow_engine.py", line 2191, in run_flow ret_val = run_flow_sync(**kwargs) ^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/prefect/flow_engine.py", line 2036, in run_flow_sync return engine.state if return_type == "state" else engine.result() ^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/prefect/flow_engine.py", line 768, in result raise self._raised File "/usr/local/lib/python3.12/site-packages/prefect/flow_engine.py", line 1287, in run_context yield self File "/usr/local/lib/python3.12/site-packages/prefect/flow_engine.py", line 2034, in run_flow_sync engine.call_flow_fn() File "/usr/local/lib/python3.12/site-packages/prefect/flow_engine.py", line 1307, in call_flow_fn result = call_with_parameters(self.flow.fn, self.parameters) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/prefect/utilities/callables/__init__.py", line 348, in call_with_parameters return fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/event_pipeline.py", line 69, in event_pipeline futures.append(global_pipeline.submit_to_work_pool( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/prefect/flows.py", line 2752, in submit_to_work_pool result = create_bundle_for_flow_run(flow=flow, flow_run=flow_run) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/prefect/bundles/__init__.py", line 510, in create_bundle_for_flow_run "function": _serialize_bundle_object(flow), ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/prefect/bundles/__init__.py", line 134, in _serialize_bundle_object return base64.b64encode(gzip.compress(cloudpickle.dumps(obj))).decode() # pyright: ignore[reportUnknownMemberType] ^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/cloudpickle/cloudpickle.py", line 1544, in dumps cp.dump(obj) File "/usr/local/lib/python3.12/site-packages/cloudpickle/cloudpickle.py", line 1313, in dump return super().dump(obj) ^^^^^^^^^^^^^^^^^ TypeError: cannot pickle '_thread._local' object
m
✅ thought for 222.7 seconds
Yep — this is a different issue now, and the traceback is actually very informative. What changed You got past flow parameter validation, so the ECS decorator itself is no longer the blocker. Now the failure happens here:
Copy code
text
global_pipeline.submit_to_work_pool(...)
...
create_bundle_for_flow_run(flow=flow, flow_run=flow_run)
...
cloudpickle.dumps(obj)
...
TypeError: cannot pickle '_thread._local' object
That means: 
submit_to_work_pool()
is trying to pickle the flow object itself, not just the parameters. So even if your parameters are now clean, something referenced by 
global_pipeline
— in its globals, closure, decorators, imported module state, or runtime context — contains a thread-local object. This is common with things like: - database/session managers - boto/session wrappers - Mongo clients - logging/context objects - thread-local config/state - module-level service singletons - 
scoped_session
- some SDK clients Important distinction 
submit_to_work_pool()
uses bundling. That means Prefect serializes the flow function and ships it to remote infrastructure. So this is stricter than normal deployment-by-source. Most likely cause in your code Even though you moved 
GlobalResources
inside the flow, your module probably still has some global object or import that pulls in thread-local state. Typical examples:
Copy code
python
mongo_client = MongoClient(...)
session = scoped_session(...)
aws_resource = SomeWrapper(...)
logger = some_custom_logger_with_context
or helpers imported from another module that create these at import time. Key clue from your traceback The object being pickled is the flow:
Copy code
text
_serialize_bundle_object(flow)
So the problem is not in the submitted arguments like 
prefect_env
or 
storage_key
. Those are fine. The problem is that 
global_pipeline
itself is capturing something unpicklable. What to do next 1. Check for module-level objects in the file containing 
global_pipeline
 Look for anything instantiated at import time, especially:
Copy code
python
client = ...
engine = ...
session = ...
resource = ...
thread_local = ...
local = threading.local()
If present, move creation of those objects inside the flow or inside tasks. Bad:
Copy code
python
mongo = MongoClient(...)
s3 = boto3.client("s3")

@flow
def global_pipeline(...):
    ...
Better:
Copy code
python
@flow
def global_pipeline(...):
    mongo = MongoClient(...)
    s3 = boto3.client("s3")
2. Check imported helper modules too Even if this file looks clean, imported modules can still define globals at import time. For example:
Copy code
python
from my_project.resources import mongo_client
If 
mongo_client
is created at import time, it can poison pickling. 3. Keep flow module “cold” at import time For flows used with 
submit_to_work_pool()
, the safest pattern is: - imports only - constants only - no live clients/resources/sessions - build clients/resources inside the flow or tasks 4. Move heavy runtime setup into tasks This is often the safest pattern:
Copy code
python
from prefect import flow, task
from prefect_aws.decorators import ecs

@task
def run_step(prefect_env: str, storage_key: str):
    import boto3
    from pymongo import MongoClient

    s3 = boto3.client("s3")
    mongo = MongoClient(...)
    return "ok"

@ecs(work_pool="my-pool")
@flow
def global_pipeline(
    prefect_env: str,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    run_step(prefect_env, storage_key)
5. Quick local debugging test In the same environment, try pickling the flow function directly:
Copy code
python
import cloudpickle
cloudpickle.dumps(global_pipeline)
or possibly:
Copy code
python
cloudpickle.dumps(global_pipeline.fn)
If that fails, it confirms the flow object/module globals are the issue. Very likely offenders Given your earlier setup, I would especially inspect for:
- boto3 sessions/resources created globally - Mongo client created globally - Athena wrappers created globally - custom resource classes with cached clients - logger adapters / context vars - anything using thread-local storage under the hood Best-practice workaround If you want reliable ECS submission via `submit_to_work_pool()`: - define the flow in a module with no runtime-initialized clients - create clients/resources only inside flow/task bodies - avoid capturing app state, framework state, or singleton objects One more strategic note If your codebase depends on lots of module-level resources, you may be better off using source-based deployments rather than 
submit_to_work_pool()
bundling. Bundling is convenient, but it’s much less forgiving about picklability. If you want, paste the top of the module containing 
global_pipeline
— imports plus any module-level variables/constants — and I can help identify the exact object that is likely causing the 
_thread._local
pickling failure.
m
@Marvin I have this to switch between the test and prod workpool? its above the flow being pickled if os.getenv("PREFECT_DEPLOY_ENV", "test") == "prod": env_switch = f"prod" else: env_switch = f"test"
m
✅ thought for 11.0 seconds
That snippet is not the problem. This is fine:
Copy code
python
if os.getenv("PREFECT_DEPLOY_ENV", "test") == "prod":
    env_switch = "prod"
else:
    env_switch = "test"
It just creates a plain string at import time, and strings are picklable. So the 
_thread._local
error is coming from something else in the module or something imported by the module. What definitely is not the issue - 
os.getenv(...)
- the 
if/else
- 
env_switch
- the formatted work pool name string What is still likely the issue Something in the file containing the flow, or one of its imports, is creating a live object at module import time, such as: - boto3 session/client/resource - Mongo client - DB engine/session - custom resource singleton - logging/context object - thread-local storage used by a library Good next debugging step In the module with 
global_pipeline
, inspect everything above the flow definition besides simple constants. Things that are safe:
Copy code
python
import os
from typing import Any, Optional

ENV = os.getenv("X", "test")
NAME = f"pool-{ENV}"
Things that are suspicious:
Copy code
python
mongo = MongoClient(...)
s3 = boto3.client("s3")
session = scoped_session(...)
resource = SomeClass(...)
Fastest isolation tactic Temporarily make a minimal flow in a brand new file:
Copy code
python
import os
from typing import Any, Optional
from prefect import flow
from prefect_aws.decorators import ecs

if os.getenv("PREFECT_DEPLOY_ENV", "test") == "prod":
    env_switch = "prod"
else:
    env_switch = "test"

@ecs(work_pool=f"mwi-data-pipelines-{env_switch}-primary")
@flow
def global_pipeline(
    prefect_env: str,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    print(prefect_env, tasks, storage_key, partition_date, other_args)
Then call:
Copy code
python
global_pipeline.submit_to_work_pool(
    prefect_env="test",
    tasks=["global_tasks"],
    storage_key="abc",
    partition_date="2026-05-20",
    other_args=None,
)
If that works, then the issue is definitely elsewhere in your real module or imports. Most useful thing you can paste next Paste: - the imports at the top of the file - any module-level variables/objects above the flow You can redact names, but keep object types/patterns. That’s the fastest way to spot the unpicklable thing.
m
@Marvin so how do I get my resources into the flow safely? Yes they are doing the mongo = MongoClient(... inside the resources process
m
✅ thought for 26.0 seconds
Safest answer: don’t pass resource/client objects into the flow at all. Pass only serializable configuration and create the resources inside the flow or tasks. Because you’re using 
submit_to_work_pool()
, Prefect bundles/pickles the flow object. If your flow’s module or closure references live clients like 
MongoClient(...)
, they can break pickling. Recommended pattern 1. Pass config, not clients Good flow params:
Copy code
python
@ecs(work_pool="my-pool")
@flow
def global_pipeline(
    prefect_env: str,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    ...
Those are safe because they’re strings/lists/dicts. 2. Build resources inside the flow/task body For example:
Copy code
python
from prefect import flow, task
from prefect_aws.decorators import ecs
from pymongo import MongoClient
import boto3
from typing import Any, Optional

def build_mongo_resource(prefect_env: str):
    uri = get_mongo_uri(prefect_env)
    return MongoClient(uri)

def build_s3_resource(prefect_env: str):
    session = boto3.Session(region_name=get_region(prefect_env))
    return session.client("s3")

@task
def do_mongo_work(prefect_env: str, storage_key: str):
    mongo = build_mongo_resource(prefect_env)
    collection = mongo["my_db"]["my_collection"]
    return collection.find_one({"storage_key": storage_key})

@task
def do_s3_work(prefect_env: str, storage_key: str):
    s3 = build_s3_resource(prefect_env)
    return s3.head_object(Bucket="my-bucket", Key=storage_key)

@ecs(work_pool="my-pool")
@flow
def global_pipeline(
    prefect_env: str,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    mongo_result = do_mongo_work(prefect_env, storage_key)
    s3_result = do_s3_work(prefect_env, storage_key)
    return {"mongo": mongo_result, "s3": s3_result}
Why this is safe - The flow parameters are serializable - The live clients are created only when the code is already running in ECS - Nothing unpicklable is captured in the bundled flow definition 3. Avoid module-level resource construction This is the main thing to avoid:
Copy code
python
mongo = MongoClient(...)
s3 = boto3.client("s3")

@flow
def global_pipeline(...):
    ...
Even if you don’t pass them as parameters, those globals can still get captured when Prefect bundles the flow. 4. Use lazy imports if needed If a library itself creates problematic state on import, move the import inside the task/function:
Copy code
python
@task
def do_mongo_work(prefect_env: str, storage_key: str):
    from pymongo import MongoClient

    mongo = MongoClient(get_mongo_uri(prefect_env))
    ...
That can help if the import path itself brings in thread-local state. 5. If you have “resource classes”, make them factories Instead of this:
Copy code
python
@dataclass
class MongoResource:
    client: MongoClient
prefer this:
Copy code
python
from dataclasses import dataclass

@dataclass
class MongoConfig:
    uri: str
    database: str

class MongoResource:
    def __init__(self, config: MongoConfig):
        self.config = config

    def get_client(self):
        from pymongo import MongoClient
        return MongoClient(self.config.uri)
Then inside the flow/task:
Copy code
python
@task
def read_from_mongo(prefect_env: str):
    config = MongoConfig(
        uri=get_mongo_uri(prefect_env),
        database="my_db",
    )
    mongo_resource = MongoResource(config)
    client = mongo_resource.get_client()
    return client[config.database]["collection"].find_one({})
Notice that the object you carry around is config, not a live connection. Good mental model Use flow parameters for: - what environment to run against - IDs / keys / dates - simple config flags - serializable dictionaries Create at runtime: - DB clients - boto clients/resources - SDK sessions - open connections - thread-local state Even better: separate config from execution
A nice structure is:
Copy code
python
def get_resource_settings(prefect_env: str) -> dict[str, Any]:
    if prefect_env == "prod":
        return {
            "mongo_uri": "...",
            "aws_region": "us-east-1",
            "bucket": "prod-bucket",
        }
    else:
        return {
            "mongo_uri": "...",
            "aws_region": "us-east-1",
            "bucket": "test-bucket",
        }

@task
def process_data(settings: dict[str, Any], storage_key: str):
    from pymongo import MongoClient
    import boto3

    mongo = MongoClient(settings["mongo_uri"])
    s3 = boto3.client("s3", region_name=settings["aws_region"])
    ...
Then the flow does:
Copy code
python
@ecs(work_pool="my-pool")
@flow
def global_pipeline(
    prefect_env: str,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    settings = get_resource_settings(prefect_env)
    process_data(settings, storage_key)
What I would change in your codebase If your “resources process” currently does something like:
Copy code
python
def get_resources(prefect_env: str) -> GlobalResources:
    mongo = MongoClient(...)
    s3 = boto3.client("s3")
    ...
    return GlobalResources(...)
that’s okay only if it is called inside the flow/task body, not at import time and not passed into the flow from outside. So this is okay:
Copy code
python
@flow
def global_pipeline(...):
    resources = get_resources(prefect_env)
    ...
But this is not okay:
Copy code
python
resources = get_resources(...)
@flow
def global_pipeline(...):
    ...
or:
Copy code
python
global_pipeline.submit_to_work_pool(global_resources=get_resources(...))
One more important nuance Even if you call 
get_resources()
inside the flow, if the module containing 
get_resources
has top-level globals like:
Copy code
python
mongo_client = MongoClient(...)
you can still get pickling failures. So make sure the resource module itself is also “cold” at import time. Best short rule - Module top level: constants, env lookups, pure helper functions only - Inside flow/task: instantiate clients/resources If you want, paste your current 
get_resources
/ resource factory code and I can rewrite it into a safe Prefect pattern.
m
inside the flow I do construct my resources @Marvin but there is a single boto client in get_resources_configuration which goes to secret manager for the secrets. Should resources = get_resources_configuration(prefect_env) global_resources = GlobalResources( athena_client=resources["athena_client"], snowflake=resources["snowflake"], mongo_data=resources["mongo_data"], s3_client=resources["s3_client"], project_resource=resources["project_resource"], pyiceberg_client=resources["pyiceberg_client"], clickhouse=resources["clickhouse"], snowflake_analytics=resources["snowflake_analytics"], slack_client=resources["slack_client"], clickhouse_analytics=resources["clickhouse_analytics"], duckdb_client=resources["duckdb_client"], ) def get_resources_configuration(prefect_env: str) -> dict[str, object]: secrets_client = boto3.client("secretsmanager", region_name="") ..... return { "dev": { "clickhouse": ClickHouseResource( host=secrets["CLICKHOUSE_HOST"], user=secrets["CLICKHOUSE_USER"], password=secrets["CLICKHOUSE_PASSWORD"], secure=True, ), @dataclass class ClickHouseResource: host: Optional[str] = None port: Optional[int] = None user: Optional[str] = None password: Optional[str] = None secure: bool = True def get_client(self) -> ch_client: return clickhouse_connect.get_client( host=self.host, port=self.port, username=self.user, password=self.password, secure=self.secure, connect_timeout=90, send_receive_timeout=600, autogenerate_session_id=False, compress=True, )
m
✅ thought for 20.1 seconds
Yes — the safest approach is: do not create even that Secrets Manager boto client at module import time. From what you described, this pattern is fine if 
get_resources_configuration(prefect_env)
is called inside the flow body or inside a task, and if the module defining it has no top-level boto clients or other live objects. So this is safe:
Copy code
python
@flow
def global_pipeline(...):
    resources = get_resources_configuration(prefect_env)
provided that 
get_resources_configuration
itself does this only at call time:
Copy code
python
def get_resources_configuration(prefect_env: str) -> dict[str, object]:
    secrets_client = boto3.client("secretsmanager", region_name="...")
    ...
That local 
secrets_client
should not cause the flow-bundling pickle error by itself, because it is created only when the flow runs. What would be unsafe If somewhere in that module you have:
Copy code
python
secrets_client = boto3.client("secretsmanager", region_name="...")
at the top level, that is a likely culprit. Or if your returned resource objects already contain live clients instead of config-only data. Based on your example, 
ClickHouseResource
looks okay if it only stores config and creates the actual client later in `get_client()`:
Copy code
python
@dataclass
class ClickHouseResource:
    host: Optional[str] = None
    ...
    def get_client(self):
        return clickhouse_connect.get_client(...)
That is a good pattern. So the main question is: Do any of the other resources in 
get_resources_configuration()
create live clients immediately? For example, these would be suspicious if they instantiate connections during config assembly: - 
MongoResource(...)
if its 
__init__
creates 
MongoClient
- 
AthenaResource(...)
if it builds boto/session objects in 
__post_init__
- 
SlackResource(...)
if it creates a 
WebClient
immediately - 
SnowflakeResource(...)
if it opens or stores a connection/session - 
DuckDBResource(...)
if it opens a DB handle immediately What is safe vs unsafe Safe resource object Stores only primitive config:
Copy code
python
@dataclass
class MongoResource:
    uri: str
    database: str

    def get_client(self):
        from pymongo import MongoClient
        return MongoClient(self.uri)
Unsafe resource object Creates client during init:
Copy code
python
@dataclass
class MongoResource:
    uri: str

    def __post_init__(self):
        self.client = MongoClient(self.uri)
That second pattern can absolutely lead to pickling trouble if the object gets captured. What I’d recommend 1. Keep 
get_resources_configuration()
pure-ish - okay to fetch secrets inside it - return config objects only - do not return active client/connection objects 2. Each resource class should lazily create clients - only in methods like 
get_client()
- not in 
__init__
, 
__post_init__
, class attributes, or module globals 3. Move secret lookup into a task if needed If bundling is still failing, one very robust pattern is to move secret retrieval into a task:
Copy code
python
from prefect import task

@task
def get_resources_configuration(prefect_env: str) -> dict[str, object]:
    import boto3

    secrets_client = boto3.client("secretsmanager", region_name="...")
    ...
    return ...
Then in the flow:
Copy code
python
@flow
def global_pipeline(...):
    resources = get_resources_configuration(prefect_env)
That can help isolate resource construction to runtime. But the critical thing here The current error is happening before the flow is launched in ECS, during bundling of the flow function. So if 
get_resources_configuration()
is only called inside the flow, then it is probably not the direct issue. That means the remaining likely causes are: - a module-level client somewhere in imports - a resource class that creates a client at import/init time - imported modules with global singletons - decorators/helpers capturing live clients
Most likely next suspect One of these resource classes: - 
AthenaResource
- 
MongoResource
- 
Boto3Resource
- 
SlackResource
- 
Snowflake...
probably creates or stores a live SDK/client object earlier than expected. Best next diagnostic step Search your resource classes for any of these patterns:
Copy code
python
client = ...
session = ...
connection = ...
MongoClient(...)
boto3.client(...)
boto3.Session(...)
clickhouse_connect.get_client(...)
snowflake.connector.connect(...)
WebClient(...)
and check whether they occur in: - module top level - class attributes - 
__init__
- 
__post_init__
If yes, that’s likely the problem. Short answer to your direct question Should you create the boto secrets client there? Yes, but only inside the function/task at runtime, like this:
Copy code
python
def get_resources_configuration(prefect_env: str) -> dict[str, object]:
    import boto3
    secrets_client = boto3.client("secretsmanager", region_name="us-east-1")
    ...
That is fine. No, if it exists at module top level. If you want, paste one or two of the other resource classes — especially 
MongoResource
and 
Boto3Resource
— because those are strong candidates for where the 
_thread._local
is being introduced.
m
@Marvin ok then it's only inside the method. and TBF when the process was running before it was pickling the entire resources object as the flow tasks on the ecs tasks were spinning up they just died with the pydantic validation error.
m
✅ thought for 18.0 seconds
That makes sense — and it points to two separate issues that you’ve been peeling back one by one: 1. First failure: Pydantic validation That was likely caused by the flow parameter schema / dataclass resource argument issue. 2. Current failure: bundling/pickling Now that the flow signature is cleaner, Prefect gets further and fails while bundling the flow object for 
submit_to_work_pool()
. So yes — earlier it may have been “working far enough” to launch ECS tasks because your resource objects were being handled differently in that path, but now the current blocker is specifically:
Copy code
text
TypeError: cannot pickle '_thread._local' object
during:
Copy code
text
create_bundle_for_flow_run(...)
cloudpickle.dumps(flow)
That means there is still something attached to the flow’s module/import graph that is not picklable. And if your secrets client is only created inside a function call, then I agree it is probably not the direct culprit. What this means practically At this point, the likely issue is not your parameters anymore and maybe not even the resource construction inside the flow body. It’s more likely one of these: - a module imported by the flow file has a top-level client/session/object - one of the decorators/utilities used by the flow references thread-local state - the flow function closes over something unpicklable indirectly - one of the imported resource classes/modules initializes library state at import time Important nuance with 
submit_to_work_pool()
 This method bundles the flow object itself. That makes it much more sensitive than source-based deployment. So even code that runs fine locally or in a normal deployment can still fail with bundling. Best next move: treat this as a bundling issue, not a resource-passing issue You have two realistic paths: Option A: Keep using 
submit_to_work_pool()
and make the flow module fully picklable This means aggressively minimizing module-level imports/state. Option B: Prefer source-based deployments instead of bundling This is often the better fit for real projects with lots of dependencies/resources. With source-based deployments, Prefect loads your code from source in the runtime environment instead of pickling the flow object and shipping it as a bundle. That usually avoids this exact class of pickling problems. My recommendation: use a normal deployment instead of 
submit_to_work_pool()
 Instead of dynamically doing:
Copy code
python
global_pipeline.submit_to_work_pool(...)
deploy the flow to the ECS work pool and then trigger runs of the deployment. For Prefect 3.x, the usual path is 
flow.deploy(...)
or 
prefect deploy
. If you want the programmatic route, it looks like:
Copy code
python
from prefect import flow

@flow
def global_pipeline(prefect_env: str, tasks: list[str], storage_key: str, partition_date: str, other_args=None):
    ...
then deploy it separately to your ECS work pool, and trigger deployment runs rather than bundling the live flow object each time. This is often much more robust. Why this helps - 
submit_to_work_pool()
-> pickles flow object - source-based deployment -> references code location/source instead That avoids the cloudpickle 
_thread._local
trap. If you want to keep 
submit_to_work_pool()
 Then the next debugging step is to isolate the exact import causing the issue. A very effective test is: 1. create a new minimal file with only: - stdlib imports - Prefect imports - the flow - no project imports 2. test 
submit_to_work_pool()
3. add imports back one by one until pickling fails That’s tedious, but it will find the offending import quickly. Why I’m leaning toward source-based deployment here Because your flow clearly depends on a richer application/resource layer. That’s normal, but it’s exactly the kind of project where bundling can be fragile. So short version - yes, these are probably two separate issues - you fixed the first one
- the current 
_thread._local
error is a bundling/pickling issue - if your boto secrets client is only created inside a function, it’s probably not the problem - best practical fix: use a source-based deployment rather than 
submit_to_work_pool()
If you want, I can show you the recommended Prefect 3.x replacement for your current 
submit_to_work_pool()
pattern using: - 
flow.from_source(...).deploy(...)
- 
prefect deploy
- creating a deployment and then triggering runs programmatically That’s probably the cleanest way forward.
m
@Marvin actually its failing sooner not getting further, it was running sub tasks they were dying so this is a new issue introduced by removing the resource object
m
✅ thought for 17.9 seconds
Got it — that distinction matters. So the timeline is: - Before: ECS subtasks were launching, then dying with the Pydantic validation error - After removing the resource object: it now fails earlier, during bundling/pickling, before ECS tasks launch So yes — this is a new issue introduced by the refactor, not just the old one surfacing differently. That strongly suggests: the way the flow is now being called or structured after the refactor introduced something unpicklable into the bundled flow path. The key line is still:
Copy code
text
global_pipeline.submit_to_work_pool(...)
...
cloudpickle.dumps(flow)
...
TypeError: cannot pickle '_thread._local' object
So the new issue is specifically about the flow object being bundled. Most likely explanation When you removed the resource object from the flow signature, you probably also changed where/how resources are created, imported, or referenced. Now the flow module or one of its helpers is capturing a live object or import-time state that cloudpickle can’t serialize. Because you said:
the process was running before it was pickling the entire resources object
that means the old code path must have been different enough that the new refactor changed what 
global_pipeline
references. What I would inspect immediately Since this is a new issue after the refactor, compare: - imports added or moved into the flow module - helper functions newly referenced by 
global_pipeline
- resource factory code now imported directly - anything now stored in defaults, closures, decorators, or module globals Specific high-probability suspects 1. A helper module imported into the flow file that has a module-level client/session 2. A resource factory function that closes over a client/session/thread-local object 3. A class imported into the module whose class attributes initialize something thread-local 4. A logging/config library with thread-local context introduced into the refactor What to test right now Take your current flow file and create a temporary minimal version in the same project:
Copy code
python
import os
from typing import Any, Optional
from prefect import flow
from prefect_aws.decorators import ecs

if os.getenv("PREFECT_DEPLOY_ENV", "test") == "prod":
    env_switch = "prod"
else:
    env_switch = "test"

@ecs(work_pool=f"mwi-data-pipelines-{env_switch}-primary")
@flow(name="global_pipeline")
def global_pipeline(
    prefect_env: str,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    return {
        "prefect_env": prefect_env,
        "tasks": tasks,
        "storage_key": storage_key,
        "partition_date": partition_date,
    }
Then call:
Copy code
python
global_pipeline.submit_to_work_pool(
    prefect_env="test",
    tasks=["global_tasks"],
    storage_key="x",
    partition_date="2026-05-20",
    other_args=None,
)
If that works, then the new issue is definitely in one of the imports or helper calls from your real flow file. Why this is the best next step Because right now we need to separate: - flow signature problems from - flow module pickling problems And your latest traceback points squarely at module/function bundling. Most likely structural fix If the refactor moved resource construction into the same module as the flow, and that module imports resource classes/factories that initialize SDK state, then: - keep the flow module extremely minimal - move heavy resource code into task-local imports or helper functions imported lazily inside the flow/task For example: ```python @ecs(work_pool=f"mwi-data-pipelines-{env_switch}-primary") @flow(name="global_pipeline") def global_pipeline( prefect_env: str, tasks: list[str], storage_key: str, partition_date: str, other_args: Optional[dict[str, Any]] = None, ): from my_project.resources import get_resources_configuration
resources = get_resources_configuration(prefect_env) ... ``` This can sometimes avoid bundling problematic imported globals at flow definition time. Short version Yes — you introduced a new bundling problem by changing the code structure. The old error was about flow parameter validation. The new one is about pickling the flow object. They are separate. Best next diagnostic action Try the minimal flow above in the same environment. If that works, then start reintroducing: 1. your original imports 2. your helper imports 3. your resource config call one at a time. If you want, paste the imports from the flow module and I can help identify which ones are most suspicious for 
_thread._local
.
m
@Marvin review this please is the way I run the tasks in the flow the issue?
Copy code
@flow(flow_run_name="global_pipeline_{storage_key}_{partition_date}", name="global_pipeline")
def global_pipeline(prefect_env: str, tasks: list[str], storage_key: str, partition_date: str, other_args: Optional[dict[str, Any]] = None):
    futures = []
    logger = get_run_logger()

    process_events_future = None


    resources = get_resources_configuration(prefect_env)
    global_resources = GlobalResources(
        athena_client=resources["athena_client"],....
    )

    <http://logger.info|logger.info>(f"Starting global tasks for date: {partition_date}")
    if "global_tasks" in tasks or "process_events" in tasks or "datalake_tasks" in tasks:
        process_events_future = process_events.submit(
            r=global_resources,
            partition_date=partition_date,
            storage_keys=[storage_key],
        )
        futures.append(process_events_future)
m
✅ thought for 20.3 seconds
Short answer: the 
.submit()
inside the flow is probably not the pickling problem. This part is normal Prefect usage:
Copy code
python
process_events_future = process_events.submit(
    r=global_resources,
    partition_date=partition_date,
    storage_keys=[storage_key],
)
Task submission inside a flow is fine. But there is still a likely problem in this snippet:
Copy code
python
resources = get_resources_configuration(prefect_env)
global_resources = GlobalResources(
    athena_client=resources["athena_client"],
    ...
)
and then:
Copy code
python
process_events.submit(r=global_resources, ...)
So while this may not be causing the flow bundling error directly, it is still a risky pattern for task execution, because you are passing a rich Python object containing resource wrappers into a task. Depending on what those resource objects contain, that can cause serialization/pickling issues at the task boundary too. So there are really two different concerns here: 1. Current error: flow bundling This happens before the flow even gets to ECS, during:
Copy code
text
global_pipeline.submit_to_work_pool(...)
That error is about serializing the flow object, not this 
process_events.submit(...)
call. So this snippet is probably not the root cause of the current 
_thread._local
error. 2. Future/runtime risk: task argument serialization Passing 
global_resources
into a task like this may still be a bad idea:
Copy code
python
process_events.submit(r=global_resources, ...)
because task inputs may also need to be serialized. What I’d change anyway Instead of passing 
global_resources
into tasks, pass only simple config and build clients/resources inside the task. For example, instead of:
Copy code
python
@task
def process_events(r: GlobalResources, partition_date: str, storage_keys: list[str]):
    client = r.clickhouse.get_client()
prefer:
Copy code
python
@task
def process_events(prefect_env: str, partition_date: str, storage_keys: list[str]):
    resources = get_resources_configuration(prefect_env)
    global_resources = GlobalResources(
        athena_client=resources["athena_client"],
        ...
    )
    client = global_resources.clickhouse.get_client()
or even better, skip 
GlobalResources
entirely inside the task and just build the exact resource needed there. So, is the task-running pattern the issue? - 
.submit(...)
itself: no, probably not - passing 
global_resources
into tasks: not ideal, and likely to bite you later - current 
_thread._local
bundling error: more likely caused by imports/module state than this task submission code One thing in your snippet to double-check This line in your pasted Slack formatting:
Copy code
python
<http://logger.info|logger.info>(f"Starting global tasks for date: {partition_date}")
I assume that’s just Slack mangling and your real code is:
Copy code
python
<http://logger.info|logger.info>(f"Starting global tasks for date: {partition_date}")
If not, fix that, but I assume it’s just formatting noise. What I suspect more than this snippet The likely issue is that 
global_pipeline
references 
get_resources_configuration
, 
GlobalResources
, and imported task functions like 
process_events
, and one of those imported modules has an unpicklable global or thread-local object. So the flow bundling problem is probably in: - the module where 
process_events
is defined - the module where 
GlobalResources
or resource classes are defined - the module where 
get_resources_configuration
is defined Recommended safer structure ```python @task def process_events(prefect_env: str, partition_date: str, storage_keys: list[str]): resources = get_resources_configuration(prefect_env) # build only what this task needs ... @flow(...) def global_pipeline(prefect_env: str, tasks: list[str], storage_key: str, partition_date: str, other_args: Optional[dict[str, Any]] = None): futures = [] logger = get_run_logger()
logger.info(f"Starting global tasks for date: {partition_date}") if "global_tasks" in tasks or "process_events" in tasks or "datalake_tasks" in tasks: futures.append( process_events.submit( prefect_env=prefect_env, partition_date=partition_date, storage_keys=[storage_key], ) ) ``` This avoids sending a composite resource object across task boundaries. My bottom line - No, this task submission pattern is probably not the cause of the current bundling error - Yes, passing 
global_resources
into tasks is something I would refactor anyway - the current 
_thread._local
issue is still more likely in imported module state than in 
.submit(...)
If you want, paste: - the imports at the top of this flow module - the signature/definition of 
process_events
- the definition of 
get_resources_configuration
and I can help pinpoint the most suspicious pickling source.
m
@Marvin as we have established this resource pattern worked fine before and didn't cause a pickling issue when being passed in so doing it inside seems unlikely to be the issue, also the methods to get_client instantiate the clients when the resources are used in tasks
m
✅ thought for 17.3 seconds
That’s a fair point — and I agree with your reasoning. If this exact resource pattern previously worked when passed into tasks, and the actual clients are only created in 
get_client()
inside the task runtime, then it’s less likely that 
global_resources
itself is the cause of the new 
_thread._local
error. So I’d update the hypothesis: the new pickling issue is probably not from passing 
global_resources
to 
process_events.submit(...)
. Because your traceback shows the failure happens before task submission matters:
Copy code
text
global_pipeline.submit_to_work_pool(...)
-> create_bundle_for_flow_run(flow=flow, flow_run=flow_run)
-> cloudpickle.dumps(flow)
That means Prefect is failing while serializing the flow object, not the task arguments. So the question becomes: what did the refactor change in the flow’s import graph or closure? Since removing the resource object from the flow signature introduced this earlier failure, the new issue is likely one of: - a new import added to the flow module - a helper now referenced directly by the flow - a decorator/wrapper around the flow/task - a module-level object in one of the imported modules - a closure over something from the caller context What is probably not the culprit - 
process_events.submit(...)
itself - lazy 
get_client()
methods - the simple 
env_switch
logic - serializable flow parameters like strings/lists/dicts What I would focus on now The actual difference between the working-before and failing-now versions. Specifically: 1. what changed in the flow signature 2. what imports were added/removed 3. whether 
submit_to_work_pool()
is now being called from a different module/context 4. whether 
get_resources_configuration
/ 
GlobalResources
/ 
process_events
are imported from modules with top-level state Best diagnostic target now Not the flow body logic — the flow module imports and the module that calls 
submit_to_work_pool()
. Because bundling the flow can fail if any referenced global in the flow’s function globals is unpicklable. Concrete thing to inspect In Python, 
cloudpickle.dumps(flow)
can fail if 
flow.fn.__globals__
contains references to problematic imported objects. That means any of these names used by the flow are suspects: - 
get_resources_configuration
- 
GlobalResources
- 
process_events
- 
get_run_logger
- any constants/helpers referenced inside the flow And if those come from modules with top-level initialized state, bundling can break. Strong next step Please paste the imports for the module containing 
global_pipeline
. Something like:
Copy code
python
import ...
from ... import ...
That is now the most useful thing to review. Because at this stage, the bug is much more likely to be caused by one imported symbol dragging in a module with thread-local state than by the task submission pattern itself. My current conclusion You’re right to push back on the resource-object theory as the primary cause of this new error. The current 
_thread._local
failure is more likely: - import-related - closure/global-related - bundling-related rather than task-argument-related. If you send the imports, I’ll review them specifically for common unpicklable patterns.
m
@Marvin only the addition on this from connectors.resources_configuration import get_resources_configuration
m
✅ thought for 19.3 seconds
That is very likely the issue. If the new failure started when you added:
Copy code
python
from connectors.resources_configuration import get_resources_configuration
then the most likely explanation is: importing 
connectors.resources_configuration
pulls in something with thread-local state at module import time. Even if 
get_resources_configuration()
itself only creates clients inside the function, the module import can still be the problem. For example, 
connectors/resources_configuration.py
might import other modules like: - Mongo helpers - boto helpers - clickhouse helpers - slack client helpers - snowflake connectors and one of those imports may initialize something globally. So yes — this new import is now my top suspect. Why this happens When Prefect bundles the flow for 
submit_to_work_pool()
, it pickles the flow function and the globals it references. By referencing 
get_resources_configuration
, the flow now depends on the imported module path behind it. If that module or its transitive imports contain a 
_thread._local
object somewhere in module state, cloudpickle can fail. Best fix to try first Instead of importing it at module level:
Copy code
python
from connectors.resources_configuration import get_resources_configuration
move the import inside the flow:
Copy code
python
@flow(flow_run_name="global_pipeline_{storage_key}_{partition_date}", name="global_pipeline")
def global_pipeline(
    prefect_env: str,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    from connectors.resources_configuration import get_resources_configuration

    futures = []
    logger = get_run_logger()

    resources = get_resources_configuration(prefect_env)
    ...
This is often enough to avoid the bundling problem, because the flow module no longer captures that imported symbol at definition time. If that still fails Then 
connectors.resources_configuration
may still be getting pulled in some other way, or another imported symbol is the issue. In that case, next step is to inspect 
connectors/resources_configuration.py
for: - top-level instantiated objects - imports of modules that instantiate objects at top level - module-level clients/sessions/resources - class-level client creation - anything using thread local/session/context machinery What to look for in that module Safe:
Copy code
python
import boto3

def get_resources_configuration(prefect_env: str):
    secrets_client = boto3.client("secretsmanager")
    ...
Suspicious:
Copy code
python
secrets_client = boto3.client("secretsmanager")
or:
Copy code
python
from connectors.mongo import mongo_client
where 
mongo_client
is created at import time. Also suspicious:
Copy code
python
slack_client = WebClient(...)
session = scoped_session(...)
client = clickhouse_connect.get_client(...)
mongo = MongoClient(...)
at module scope anywhere in that import chain. Most likely practical outcome I would try this in order: 1. Move the import inside the flow 2. If needed, move it even further inside the task that actually needs it 3. Audit 
connectors.resources_configuration
and its imports for module-level clients So yes: I think you found the likely trigger. The new import:
Copy code
python
from connectors.resources_configuration import get_resources_configuration
is the strongest suspect for the 
_thread._local
pickling failure. If you want, paste 
connectors/resources_configuration.py
imports and top-level code, and I’ll help identify the specific problematic import.
m
@Marvin it fails in your code here with _pickle_local_modules_by_value(flow): bundle: SerializedBundle = { "function": _serialize_bundle_object(flow),
m
✅ thought for 17.2 seconds
Yes — that’s consistent with everything we’ve narrowed down. That line means Prefect is failing while serializing the flow object itself:
Copy code
python
with _pickle_local_modules_by_value(flow):
    bundle = {
        "function": _serialize_bundle_object(flow),
    }
So the issue is definitely in something reachable from the flow object: - its function globals - referenced imports - closure - decorators/wrappers - module state from imported symbols Not in the runtime execution of the flow body yet. What 
_pickle_local_modules_by_value(flow)
implies Prefect is trying to register local modules used by the flow so they can be bundled by value. During that process, 
cloudpickle
walks the flow and its references. If one of those references leads to a 
_thread._local
, pickling fails immediately. So this confirms: - not an ECS networking issue - not task execution yet - not a runtime 
get_client()
call yet - not the submitted parameters - yes: a bundling/import graph problem Given your latest info, the strongest suspect remains:
Copy code
python
from connectors.resources_configuration import get_resources_configuration
or something imported by that module. Because that was the meaningful change before this new error appeared. What to do next The best immediate mitigation is to remove that symbol from the flow’s global namespace. Instead of this at module level:
Copy code
python
from connectors.resources_configuration import get_resources_configuration
use an inner import:
Copy code
python
@flow(flow_run_name="global_pipeline_{storage_key}_{partition_date}", name="global_pipeline")
def global_pipeline(
    prefect_env: str,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    from connectors.resources_configuration import get_resources_configuration
    ...
That reduces what the flow object directly references at pickle time. Why this may help 
cloudpickle
sees names used by the function. If a problematic object is attached to a module-level import path, deferring the import can keep it out of the global references captured in the serialized flow. It’s not guaranteed, but it’s often effective. If inner import still fails Then one of these is true: 1. another global import is the real culprit 2. 
process_events
or another referenced task pulls in the same problematic module 3. the module is considered local and still bundled through another reference In that case, the next step is to inspect the imported modules used by: - 
global_pipeline
- 
process_events
especially any local project modules. High-signal debugging suggestion Run a quick binary isolation test by commenting out pieces referenced in the flow body: 1. keep only logger + return 2. add 
process_events.submit(...)
3. add the inner import 4. add the call to 
get_resources_configuration
5. add 
GlobalResources(...)
This helps identify which symbol reference causes the bundle serialization to fail. Example minimal progression:
Copy code
python
@flow
def global_pipeline(...):
    logger = get_run_logger()
    <http://logger.info|logger.info>("hello")
    return
then:
Copy code
python
@flow
def global_pipeline(...):
    from connectors.resources_configuration import get_resources_configuration
    logger = get_run_logger()
    <http://logger.info|logger.info>("hello")
    return
If step 1 works and step 2 fails, you’ve confirmed the import is enough by itself. Bottom line That traceback confirms this is a flow serialization problem, and the likely trigger is still one of the imported local modules referenced by the flow. If you want, paste: - the imports of 
connectors.resources_configuration
- any top-level code in that module - the import lines for 
process_events
and I’ll help spot the likely 
_thread._local
source.
m
should I move all the modules into the flow?
@Marvin should I just move all the modules into the flow?
m
✅ thought for 13.8 seconds
Not all of them. Don’t move everything inside the flow by default. That gets messy fast. Instead, move only the suspicious local/project imports that may be pulling in runtime state. Good rule of thumb Keep these at module level: - standard library imports (
os
, 
typing
, 
datetime
, etc.) - Prefect imports - simple third-party libraries that don’t create runtime state on import Consider moving these inside the flow/task: - your own project modules, especially 
connectors.*
, 
resources.*
, 
clients.*
- anything that may import SDK clients, DB/session code, app config, logging context, etc. So for your case, I would specifically try moving imports like:
Copy code
python
from connectors.resources_configuration import get_resources_configuration
from ...
inside the flow or task. Example:
Copy code
python
from typing import Any, Optional
from prefect import flow, get_run_logger
from prefect_aws.decorators import ecs

@ecs(work_pool=f"mwi-data-pipelines-{env_switch}-primary")
@flow(flow_run_name="global_pipeline_{storage_key}_{partition_date}", name="global_pipeline")
def global_pipeline(
    prefect_env: str,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    from connectors.resources_configuration import get_resources_configuration
    from my_project.models import GlobalResources
    from my_project.tasks import process_events

    futures = []
    logger = get_run_logger()

    resources = get_resources_configuration(prefect_env)
    global_resources = GlobalResources(
        athena_client=resources["athena_client"],
        ...
    )

    <http://logger.info|logger.info>(f"Starting global tasks for date: {partition_date}")

    if "global_tasks" in tasks or "process_events" in tasks or "datalake_tasks" in tasks:
        futures.append(
            process_events.submit(
                r=global_resources,
                partition_date=partition_date,
                storage_keys=[storage_key],
            )
        )
Why this helps It keeps the flow object’s global namespace smaller and cleaner during bundling. What not to move unless needed You usually do not need to move: - 
from prefect import flow, task, get_run_logger
- 
from typing import Any, Optional
- 
import os
Those are not likely to cause 
_thread._local
issues. Best strategy Don’t do a huge refactor all at once. Do it incrementally: 1. Move 
from connectors.resources_configuration import get_resources_configuration
inside the flow 2. Test 3. If still failing, move imports for local tasks/models used by the flow inside too 4. If still failing, inspect those imported modules for top-level client creation One caution If you move task definitions themselves into the flow, that’s usually not what you want. Move the imports of task functions if needed, but don’t redefine tasks inside the flow unless you have a strong reason. So: Okay
Copy code
python
def global_pipeline(...):
    from my_project.tasks import process_events
Not ideal
Copy code
python
def global_pipeline(...):
    @task
    def process_events(...):
        ...
My recommendation for your case No, don’t move all modules. Move your local connector/resource/task imports first, especially the newly introduced ones. If you want, I can help you make a short “safe at top-level vs move inside flow” list based on your actual imports.
m
@Marvin ok done
Copy code
cannot pickle '_thread._local' object
  File "/Users/mathewpartridge/Projects/mwi-data-pipelines/prefect_flows/event_pipeline.py", line 69, in event_pipeline
            prefect_env=prefect_env,
                       ^^^^^^^^^^^^^
    ...<5 lines>...
            partition_date=partition_date,

  File "/Users/mathewpartridge/Projects/mwi-data-pipelines/prefect_flows/event_pipeline.py", line 90, in <module>
    event_pipeline(env_override="dev", tasks= ["clickhouse_schemas","process_agentic_campaign_revenue"], storage_key="", partition_date=dates)
    ~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: cannot pickle '_thread._local' object
m
✅ thought for 12.1 seconds
That traceback is a very useful clue. The key detail is this:
Copy code
text
File ".../event_pipeline.py", line 90, in <module>
    event_pipeline(...)
That means the problem may be tied to the fact that you are calling the parent flow at module import/runtime in the same file, not just defining it. So you have something like:
Copy code
python
event_pipeline(
    env_override="dev",
    tasks=["clickhouse_schemas", "process_agentic_campaign_revenue"],
    storage_key="",
    partition_date=dates,
)
at the bottom of the module. Why that matters When the module is imported/executed, that top-level flow invocation can create Prefect runtime context and objects that are not picklable. Then when 
submit_to_work_pool()
runs inside that flow, Prefect tries to bundle another flow and hits 
_thread._local
. This is especially suspicious because the error points both to: - the line where 
submit_to_work_pool()
is called - and the top-level module call to 
event_pipeline(...)
So yes: this may be the issue. What you should do Wrap the direct invocation in a main guard:
Copy code
python
if __name__ == "__main__":
    event_pipeline(
        env_override="dev",
        tasks=["clickhouse_schemas", "process_agentic_campaign_revenue"],
        storage_key="",
        partition_date=dates,
    )
This is important. Without that guard, the module executes the flow call as soon as Python imports/runs the file, which can interfere with bundling and create exactly the kind of weird serialization behavior you’re seeing. Why this can fix it It prevents the flow from being executed at import time when the module is being analyzed/serialized. Recommended structure
Copy code
python
from prefect import flow

@flow
def event_pipeline(...):
    ...

if __name__ == "__main__":
    dates = "2026-05-20"
    event_pipeline(
        env_override="dev",
        tasks=["clickhouse_schemas", "process_agentic_campaign_revenue"],
        storage_key="",
        partition_date=dates,
    )
Very important related note For any file containing flows, avoid top-level code that: - runs flows - creates clients - loads runtime state - performs work other than definitions/constants Top-level should ideally just be: - imports - constants - function/task/flow definitions My confidence here Pretty high that this is at least part of the problem, because the traceback explicitly points to the module-level invocation. Next step Please try adding the 
if __name__ == "__main__":
guard and rerun. If it still fails after that, then the next suspects are: - module-level imports in 
event_pipeline.py
- imported task modules referenced by 
event_pipeline
- imported modules used by the child flow being submitted But the main-guard fix is the first thing I’d do. Docs: - Prefect docs
m
@Marvin no I am just running it locally to investigate its the same error and yes its name == main'd
m
✅ thought for 23.1 seconds
Got it — thanks, that helps. If it’s already under 
if __name__ == "__main__":
, then the traceback isn’t saying the module-level call is the cause; it’s just showing the local entrypoint where the failure surfaced. So at this point, the important fact is: calling 
global_pipeline.submit_to_work_pool(...)
is still forcing Prefect to bundle/pickle the 
global_pipeline
flow object, and something referenced by that flow is unpicklable. And since you already moved the suspicious import(s) inward, the remaining likely causes are now: 1. another imported local module referenced by the flow 2. the task function being submitted inside the flow 3. the 
@ecs
-wrapped child flow itself 4. Prefect context interacting badly with local submission while inside another flow The strongest suspect now is actually this pattern itself:
Copy code
python
futures.append(global_pipeline.submit_to_work_pool(...))
inside another running flow. Because 
submit_to_work_pool()
bundles the child flow object while you are already inside Prefect runtime context. That can expose context/thread-local state via globals or wrappers more easily than a plain script call. Important distinction This is not the same as 
task.submit()
. 
submit_to_work_pool()
is serializing a flow object for remote execution, and that’s a heavier, stricter path. What I think is most likely now The object being pickled is probably the 
@ecs
-decorated flow object plus its Prefect/runtime wrapper state, not just your own resources. You’ve stripped away a lot of the obvious custom suspects already. So the remaining issue may be a limitation/bug in using 
submit_to_work_pool()
from inside another flow in your current setup. Best way to test that hypothesis Try calling the child flow submission from a plain Python script, outside of any parent flow. For example:
Copy code
python
from my_module import global_pipeline

if __name__ == "__main__":
    global_pipeline.submit_to_work_pool(
        prefect_env="dev",
        tasks=["global_tasks"],
        storage_key="x",
        partition_date="2026-05-20",
        other_args=None,
    )
If that works, but calling it from inside 
event_pipeline
fails, then the issue is specifically the nested flow submission context. If that fails too Then the child flow object itself is still not picklable, and we need to isolate which symbol reference on that flow causes it. Most useful isolation step now Make a brand-new minimal child flow in the same file and submit that to the work pool:
Copy code
python
from typing import Optional, Any
from prefect import flow
from prefect_aws.decorators import ecs

@ecs(work_pool="your-pool")
@flow(name="test_child_flow")
def test_child_flow(
    prefect_env: str,
    tasks: list[str],
    storage_key: str,
    partition_date: str,
    other_args: Optional[dict[str, Any]] = None,
):
    return "ok"
Then from your parent flow or plain script:
Copy code
python
test_child_flow.submit_to_work_pool(
    prefect_env="dev",
    tasks=["x"],
    storage_key="x",
    partition_date="2026-05-20",
    other_args=None,
)
This test is extremely valuable. - If minimal child flow works: your real 
global_pipeline
still references something unpicklable - If minimal child flow also fails: the issue is likely with 
submit_to_work_pool()
+ 
@ecs
+ current runtime/version context My recommendation now Do these in order: 1. test a minimal child flow with 
@ecs
and no project imports 2. test that child flow from a plain script, outside any parent flow 3. if the minimal version works, reintroduce imports one by one Practical conclusion You’ve already ruled out a lot of your resource code. At this point the most likely problem is either: - a remaining symbol referenced by 
global_pipeline
, or - the 
submit_to_work_pool()
bundling path itself in this nested-flow usage
If you want, I can help you write a 2-minute “minimal reproduction matrix” to pinpoint exactly which combination breaks: - plain script vs parent flow - bare flow vs real flow - 
@flow
vs 
@ecs
+ 
@flow
m
@Marvin I added the below at the start of my main pipeline before anything is called
Copy code
Exception has occurred: TypeError
cannot pickle '_thread._local' object
  File "/Users/mathewpartridge/Projects/mwi-data-pipelines/prefect_flows/event_pipeline.py", line 30, in event_pipeline
    global_pdasdipeline.submit_to_work_pool()
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^
  File "/Users/mathewpartridge/Projects/mwi-data-pipelines/prefect_flows/event_pipeline.py", line 100, in <module>
    event_pipeline(env_override="dev", tasks= ["clickhouse_schemas","process_agentic_campaign_revenue"], storage_key="", partition_date=dates)
    ~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: cannot pickle '_thread._local' object
@ecs( work_pool=f"", ) @flow(flow_run_name="global_pipeline", name="global_pipeline") def global_pdasdipeline(): print(f"Starting global tasks for date:") global_pdasdipeline.submit_to_work_pool()
m
✅ thought for 16.2 seconds
Thanks — this is the clearest signal yet. If this minimal flow still fails:
Copy code
python
@ecs(work_pool="...")
@flow(flow_run_name="global_pipeline", name="global_pipeline")
def global_pdasdipeline():
    print("Starting global tasks for date:")

global_pdasdipeline.submit_to_work_pool()
then your resource code is not the cause of the current pickling failure. That means the issue is much more fundamental: either 
submit_to_work_pool()
+ 
@ecs
is not working in your current environment, or the surrounding Prefect runtime context is making the flow object unpicklable. Given what you just showed, the strongest conclusion is: the problem is with bundling an 
@ecs
infrastructure-bound flow via 
submit_to_work_pool()
in your current setup. And since your earlier traceback showed Python 3.12 in the runtime container, while your local machine path is macOS and likely a newer Python locally, environment/version behavior may also be involved. What this test proved You stripped away: - custom parameters - resource objects - task submission - helper imports - local project modules and it still fails. So the bug is now very unlikely to be in your application code. Most likely remaining causes 1. A Prefect bug/limitation with 
submit_to_work_pool()
on infrastructure-bound flows 2. A version/environment incompatibility in the local runtime 3. A known bundling/pickling edge case with the 
@ecs
decorator path Given the evidence, I would stop trying to “fix” your resource code for this specific error. What I recommend now Do not use 
submit_to_work_pool()
for this flow right now. Instead, use a normal deployment to the ECS work pool and trigger runs from the deployment. That avoids the bundling path that is blowing up here. Preferred alternatives Option 1: deploy the flow to the ECS work pool, then trigger the deployment In Prefect 3.x, the recommended deployment flow is either: - 
prefect deploy
- or programmatic deployment methods I should verify the exact CLI syntax before recommending commands, but conceptually the pattern is: - create a deployment for 
global_pipeline
- target your ECS work pool - trigger runs of that deployment instead of 
submit_to_work_pool()
This uses source/deployment-based execution instead of pickling the live flow object. Why this is the right move Your minimal reproduction strongly suggests the issue is not in your business logic. Continuing to debug your resources won’t fix a serialization bug in the bundling path. What not to use - don’t keep chasing 
GlobalResources
for this specific error - don’t keep moving imports around hoping the minimal flow will suddenly pickle You already proved the minimal example fails. Bottom line Your latest test indicates: 
submit_to_work_pool()
itself is the failing path for your 
@ecs
flow in this environment. So the practical workaround is: - use a deployment on the ECS work pool - trigger deployment runs instead of calling 
.submit_to_work_pool()
If you want, I can help you convert this exact pattern into a Prefect 3.x deployment-based approach and verify the exact commands/API syntax for: - 
prefect deploy
- creating the deployment against your ECS work pool - triggering runs programmatically afterward
m
@Marvin it was the parent flow, moved the imports from the parent flow in the parent flow def and it resolved. New question, how do I control concurrency on those sub flows?
m
✅ thought for 19.9 seconds
Yes — that tracks. Glad you found it. For controlling concurrency on the subflows, there are a few different levers depending on what kind of concurrency you mean: 1. Limit how many subflows you submit at once in the parent flow If you’re doing something like:
Copy code
python
futures.append(child_flow.submit_to_work_pool(...))
for many child flows, you can gate how many are launched concurrently in Python. A simple batching pattern:
Copy code
python
from prefect import flow

@flow
def parent_flow(items: list[str], batch_size: int = 5):
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        futures = []

        for item in batch:
            futures.append(
                child_flow.submit_to_work_pool(
                    storage_key=item,
                )
            )

        for future in futures:
            future.wait()
That limits concurrent child flow launches to 
batch_size
. 2. Limit concurrency at the work pool / worker level If all subflows go to an ECS work pool, concurrency is also constrained by: - number of worker processes - AWS/ECS capacity - work pool queueing behavior - any pool-level limits you’ve configured This is more infrastructure-driven than Prefect-driven. 3. Use Prefect global concurrency limits If you want a real Prefect-managed concurrency cap across runs, use a global concurrency limit. This is usually the best answer if you want to say: - only 3 of these child flows may run at a time - even across multiple parent flow runs You’d typically create a global concurrency limit and have the subflows acquire it. In Prefect 3.x, concurrency limits are documented here: Global concurrency limits 4. Tag-based concurrency If your concern is limiting tasks/flows by logical category, Prefect also has tag-based concurrency controls: Task run concurrency limits But for subflows submitted to work pools, global concurrency limits are usually the cleaner fit. Practical recommendation for your case If you want to limit the number of ECS child flows launched from a parent flow, I’d use one of these: - simple and explicit: batch them in the parent flow - centralized and robust: global concurrency limits Simple batching example
Copy code
python
@flow
def event_pipeline(storage_keys: list[str], prefect_env: str):
    batch_size = 3

    for i in range(0, len(storage_keys), batch_size):
        batch = storage_keys[i:i + batch_size]
        futures = []

        for storage_key in batch:
            futures.append(
                global_pipeline.submit_to_work_pool(
                    prefect_env=prefect_env,
                    tasks=["global_tasks"],
                    storage_key=storage_key,
                    partition_date="2026-05-20",
                )
            )

        for future in futures:
            future.wait()
That ensures only 3 subflows are in flight from this parent at one time. If you want concurrency across all parent runs too Then use a global concurrency limit rather than batching alone. One caveat If you mean inside each child flow rather than among child flows, that’s a different question: - task runner controls parallelism inside a flow - concurrency limits control cross-run/task coordination So if you want, I can show you either: 1. limit the number of child flows submitted from the parent 2. limit the number of child flow runs globally in Prefect 3. limit task concurrency inside each child flow If you want the exact CLI/UI steps for global concurrency limits, I can verify the current Prefect 3.x command syntax first and give you the exact command.
3 Views
Open in Slack
PreviousNext
Powered by