Hello, I am getting an error when I run a deployment using an ecs block. There are no logs in Prefect for the flow run, just says error, but if I look at the logs of the agent in cloudwatch, I see "h2.exceptions.Protocol*Error*: Invalid input ConnectionInputs.RECV_PING in state ConnectionState.CLOSED". What does this mean?

This message was deleted.

Hello, I'm using prefect 1 with kubernetes agents, so far i used to launch through

prefect agent kubernetes start

and i am trying to migrate to

prefect agent kubernetes install

to generate a yaml file. I use the --job-template option which is not available with agent *install*, is there a workaround ?

We have a few tasks that are hanging and taking the concurrency limit, but their flow has been cancelled. How do we cancel those tasks?

Prefect Version: 2.7.0 Agent: Azure Kubernetes Problem: We just experienced an issue with our Prefect Agent where it crashed. Two flow runs with the same work queue where the concurrency limit is 5 were submitted. These flow runs trigger every 5 minutes. Then, about 15 seconds later, the k8s pod running the Prefect agent crashed. K8s created a new agent pod automatically and our flow runs resumed back to normal. The K8s resource consumption looks normal: CPU Usage 5%, Memory Usage 10%. This issue has occurred about once every two weeks for the last month or so. Between the crashes, everything on the k8s cluster (Prefect agent and Prefect flow runs as k8s jobs) runs normally. Traceback from the agent pod:

13:39:09.646 | INFO    | prefect.agent - Submitting flow run '64c00c09-07a6-495b-ada4-a4088e122434'
13:39:09.647 | INFO    | prefect.agent - Submitting flow run 'beb26192-33fb-4988-985b-b497c35573f0'
13:39:10.205 | INFO    | prefect.infrastructure.kubernetes-job - Job 'bright-leech-f77cl': Pod has status 'Pending'.
13:39:10.208 | INFO    | prefect.infrastructure.kubernetes-job - Job 'gleaming-griffin-w55qm': Pod has status 'Pending'.
13:39:10.240 | INFO    | prefect.agent - Completed submission of flow run '64c00c09-07a6-495b-ada4-a4088e122434'
13:39:10.245 | INFO    | prefect.agent - Completed submission of flow run 'beb26192-33fb-4988-985b-b497c35573f0'
13:39:11.736 | INFO    | prefect.infrastructure.kubernetes-job - Job 'bright-leech-f77cl': Pod has status 'Running'.
13:39:11.912 | INFO    | prefect.infrastructure.kubernetes-job - Job 'gleaming-griffin-w55qm': Pod has status 'Running'.
13:39:24.462 | ERROR   | prefect.agent - An error occured while monitoring flow run '92139cd4-23df-4396-9150-9804a17968d6'. The flow run will not be marked as failed, but an issue may have occurred.
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/urllib3/response.py", line 761, in _update_chunk_length
    self.chunk_left = int(line, 16)
ValueError: invalid literal for int() with base 16: b''
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/urllib3/response.py", line 444, in _error_catcher
    yield
  File "/usr/local/lib/python3.10/site-packages/urllib3/response.py", line 828, in read_chunked
    self._update_chunk_length()
  File "/usr/local/lib/python3.10/site-packages/urllib3/response.py", line 765, in _update_chunk_length
    raise InvalidChunkLength(self, line)
urllib3.exceptions.InvalidChunkLength: InvalidChunkLength(got length b'', 0 bytes read)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/agent.py", line 417, in _submit_run_and_capture_errors
    result = await infrastructure.run(task_status=task_status)
  File "/usr/local/lib/python3.10/site-packages/prefect/infrastructure/kubernetes.py", line 277, in run
    return await run_sync_in_worker_thread(self._watch_job, job_name)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 69, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/usr/local/lib/python3.10/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/usr/local/lib/python3.10/site-packages/prefect/infrastructure/kubernetes.py", line 527, in _watch_job
    for log in logs.stream():
  File "/usr/local/lib/python3.10/site-packages/urllib3/response.py", line 624, in stream
    for line in self.read_chunked(amt, decode_content=decode_content):
  File "/usr/local/lib/python3.10/site-packages/urllib3/response.py", line 816, in read_chunked
    with self._error_catcher():
  File "/usr/local/lib/python3.10/contextlib.py", line 153, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/usr/local/lib/python3.10/site-packages/urllib3/response.py", line 461, in _error_catcher
    raise ProtocolError("Connection broken: %r" % e, e)

urllib3.exceptions.ProtocolError: ("Connection broken: InvalidChunkLength(got length b'', 0 bytes read)", InvalidChunkLength(got length b'', 0 bytes read))
13:39:29.507 | ERROR   | prefect.infrastructure.kubernetes-job - Job 'gleaming-griffin-w55qm': Job did not complete.
13:39:29.511 | ERROR   | prefect.infrastructure.kubernetes-job - Job 'bright-leech-f77cl': Job did not complete.

hey all, in prefect 1, for a flow of flows, how do you cancel all flows at once? i am canceling the parent-most flow, and the child flows still remain live

hey all, I am trying to understand the intended use of `any_successful`: I have a task I want to run when any upstream task succeeds, and to then get the value from it to use. It's not clear to me, though, how to go about finding which

upstream_task

succeeded. Am I on the right track? I've been poking around for a tutorial or example, but am having trouble finding much in the way of examples of

any_successful

@task(trigger=any_successful)
def bar(upstream_tasks):
    # get the first upstream_task that succeeded

gratitude thank you in advance for any insights you might have

I'm looking at converting some older prefect 0.X.X version flows to 2.X.X, 1 use case that I'm looking at uses multiple schedule with separate

parameter_defaults

(example below)

abc = IntervalClock(
    start_date=pendulum.datetime(2022, 1, 12, 3, 30, 0, tz="America/New_York"),
    interval=datetime.timedelta(hours=24),
    parameter_defaults={"mart": "abc"},
)

xyz = IntervalClock(
    start_date=pendulum.datetime(2022, 1, 12, 3, 30, 0, tz="America/New_York"),
    interval=datetime.timedelta(hours=24),
    parameter_defaults={"mart": "xyz"},
)

Is there a way have a flow use multiple schedules + to pass

parameter_defaults

to a schedule in prefect 2.0?

Prefect 2.6.8 -> I'm getting

AttributeError: 'unmapped' object has no attribute 'get'

when using a

.map()

on a task. I'm passing a

dict

. Upstream flows that unmap this object does not throw any error. Any thoughts on what a solution could be?

Hi everyone, We are getting started with Prefect Orion and are excited about the possibilities. Our central use case is coordinating lots of data jobs that integrate into one master database. We'd like to set up a continuous integration (CI) environment to ensure that when we create or rev Flows that these Flows are fully integration/regression tested. Our current idea (open to other approaches) is to set things up so that when we go to put a new Flow in CI all the Flows (including dependents) can be fired against the CI database and not the production database. It would seem like this requires some way to either maintain separate Flows or set a parameter that goes through all flows that indicates whether this is a dev run, a CI run, or a production run. Has anyone done anything like this? If so, how did you accomplish it? Any other ideas/feedback welcome. Thanks!

Hi, currently running Prefect 2.4.0 on GCP GKE autopilot. I keep getting intermittent flow crashes like this.

Is there a way to rollback a flow to a specific version in prefect 2?

Prefect version: 2.7.8 Agent: Azure Kubernetes Problem: I just upgraded to Prefect 2.7.8 and canceled a flow run through the UI. The flow canceled as expected, however in the agent logs I see the following:

20:20:14.675 | INFO    | prefect.agent - Found 1 flow runs awaiting cancellation.
20:20:14.675 | ERROR   | prefect.agent - Flow run 'da1b2a10-0a7a-48a1-98c3-25f92f3be5a8' does not have an infrastructure pid attached. Cancellation cannot be guaranteed.

What do I need to do to attach a PID to the flow object? Also important to note, I had to add a cluster role to my K8s Prefect service account to get Prefect 2.7.8 to work. I couldn't find this requirement outlined anywhere in the docs. I could have just missed it. https://github.com/PrefectHQ/prefect/issues/7842.

Couple of questions: 1. can run_deployment be used inside a task? <- specifically a mapped task 2. if so, how can wait until all the flow runs are successful?

Hi for the prefect agent: is there a way to add an additional header for HTTP requests to the prefect-cloud or s3 or git hub?

I’m having a hard time understanding task runners vs. async tasks and subflows. Why would one use

ConcurrentTaskRunner

/`DaskTaskRunner` /`RayTaskRunner` vs. using some combination of async tasks and/or async subflows? Any tips?

Is there a way to use azure application insights to work with prefect flow/task logging?

Hi prefect team For one of my flow i got this error (when using prefect-dbt

trigger_dbt_cli_command()

)

prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

The weird thing is • I didn’t restart the job it a basic scheduled jobs • It worked well before upgrading to 2.7.6 • I added the flag

PREFECT_RESULTS_PERSIST_BY_DEFAULT: true

but no effect

Am I correct in understanding that currently there is no way to execute subflows as if they were tasks with a task runner? For example, if I want to run three subflows in parallel with Ray that isn’t currently possible right?

Hi 👋 I am trying to define my own new Infrastructure class, but getting the error: “No class found for dispatch key ‘xxx’ in registry for type ‘Block’” full info in this thread: https://discourse.prefect.io/t/submission-failed-keyerror-no-class-found-for-dispatch-key-xxx-in-registry-for-type-block/2203/1 Any ideas what am I doing wrong here?

In case that a prefect flow crashes with: "Execution was interrupted by an unexpected exception.", what can be done? I have added retries to the flow but it is not picked up.

Hi, can anyone enlighten me regarding scaling prefect The scenario is i have setup my inferencing pipeline where the user upload an image goes through the pipeline and return back the results The problem here is there can be 1000-2000 inferencing request can come in one go, and then the question is 1. Will prefect can handle it 2. Is yes then how (my prefect is setup on cloud) Is there something like i have to setup auto-scaling kindof thing Please if anyone can help me on this

Hi, I run into this error but I can not fix it. The problem is: it run ok on my VSCode, but when I create a deployment and start an agent and run the code from prefect UI, it keep showing this error in Flows logs.

UnicodeEncodeError: 'charmap' codec can't encode characters in position 13-14: character maps to <undefined>

I think it would be the same problem with this one: https://github.com/PrefectHQ/prefect/issues/5754 Thank you for your help!

Encountered exception during execution:
Traceback (most recent call last):
  File "D:\python\lib\site-packages\prefect\engine.py", line 1445, in orchestrate_task_run
    result = await run_sync(task.fn, *args, **kwargs)
  File "D:\python\lib\site-packages\prefect\utilities\asyncutils.py", line 154, in run_sync_in_interruptible_worker_thread
    async with anyio.create_task_group() as tg:
  File "D:\python\lib\site-packages\anyio\_backends\_asyncio.py", line 662, in __aexit__
    raise exceptions[0]
  File "D:\python\lib\site-packages\anyio\to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "D:\python\lib\site-packages\anyio\_backends\_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "D:\python\lib\site-packages\anyio\_backends\_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "D:\python\lib\site-packages\prefect\utilities\asyncutils.py", line 135, in capture_worker_thread_and_result
    result = __fn(*args, **kwargs)
  File "C:\Users\WILLTR~1\AppData\Local\Temp\tmpdl74duviprefect\prefect_test.py", line 106, in st1_extract_order_data
    print('Client: ' + shop_name)
  File "D:\python\lib\encodings\cp1252.py", line 19, in encode
    return codecs.charmap_encode(input,self.errors,encoding_table)[0]
UnicodeEncodeError: 'charmap' codec can't encode characters in position 13-14: character maps to <undefined>

Hi all - a question. Say I have a flow run that has a task runner, and in said flow run I have a sub-flow I execute. Is there a way to inherit the task runner from the parent flow? I ask because my task runner is backed by a slurm job allocation managed via

dask_jobqueue

. It makes sense to me to avoid a whole new set of requests when resources have already been allocated for the blocked parent flow.

Hey guys, is it possible to change a schedule start time of a flow run from graphql in prefect 1.2 ?

Heya! Is anyone using P Prefect with Ⓜ️ Meltano, dbt dbt? As in • Meltano for the EL • dbt for the T Then Prefect for the orchestration, monitoring and generally to glue the ELT together I'm curious what would be the preferred, best practice in plugging in Meltano in Prefect? I.e. container the meltano project and then run the container as a task in Prefect.. or something else? There seem to not be much of any blogs on the matter 🤷

Hi prefect team For one of my flow i got this error (when using prefect-dbt

trigger_dbt_cli_command()

)

prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

The weird thing is • I didn’t restart the job it a basic scheduled jobs • It worked well before upgrading to 2.7.6 • I added the flag

PREFECT_RESULTS_PERSIST_BY_DEFAULT: true

but no effect

Hi All, I'm new to prefect and try to build the following. I have some Code on Bitbucket. I create a Bitbucket-Storage-Block and create a deployment which fetchs a flow from Bitbucket and run it locally. On Bitbucket there are also some test-files which I try to read. So inside my Flow I pull this specific folder with block.get_directory() so that the date persists locally. Each time I trigger this flow, the folder is re-pulled from Bitbucket. Is there the possibilty that the pull is executed only if the files on the specifiy Bitbucket-Path are newer than the previous pulled local state?

Hi - I’m trying to use GitHub storage block (private repo with access token) in my deployment using AWS k8s backend, but I’m not sure where the AWS credentials need to be passed (deployment succeeds but submitting a flow run returns 401)

github_block = GitHub.load("XXXXXX")

cluster_config_block = KubernetesClusterConfig.load("data-engineering")

k8s = KubernetesJob(
    cluster_config=cluster_config_block
)

deployment = Deployment.build_from_flow(
    flow=test,
    name="test",
    version="0.0.1",
    tags={"from-python-object"},
    parameters={"name": "test-run"},
    infra_overrides={"env": {"PREFECT_LOGGING_LEVEL": "DEBUG"}},
    work_queue_name="test",
    infrastructure=k8s,
    storage=github_block,
)

if __name__ == "__main__":
    result = deployment.apply()

Hello everyone, i am getting following warning

WARNING: prefect 2.0.0 does not provide the extra 'aws'
WARNING: prefect 2.0.0 does not provide the extra 'google'
WARNING: prefect 2.0.0 does not provide the extra 'snowflake'
WARNING: prefect 2.0.0 does not provide the extra 'viz'

When doing :

pip install "prefect[aws,google,snowflake,viz]==2.0.0"

am i doing something wrong ?