Hi all 🖐️ I'm having an issue with an agent connected to Prefect Cloud. The agent is listening to a queue named _agent_1._ The concurrency limit for the queue is 4. Everything worked as expected for many weeks, but recently the agent stopped receiving new runs. The it started receiving runs again after I increased the concurrency limit to 100. I know that sometimes this happens when there are "ghost" runs - runs in Pending or Running state that are not really being executed by an agent, but they contribute to "used" concurrency number. But this time it is different. There are no Pending or Running runs and the total count of runs in non-finished state (mostly crashed) is about 20. But processing starts only when I set the concurrency limit to a value that is way higher than 20 (above 100). My question is, what causes this and how can it be fixed and avoided? Apparently, Orion/Prefect Cloud thinks the concurrency limit is reached, but I don't understand why. Any help greatly appreciated 🙏

Hello 👋 I have a self-hosted Prefect (deployed with helm on k8s), is it possible to connect to the Orion API for example from GitHub Actions to deploy flows from the CI? Thanks!

Hi, all! Running Prefect 2, created a Docker block, created and applied deployment (successful runs NOT using docker), and when I run the docker deployment, I get: httpx.ConnectError: All connection attempts failed I've set the env variable of the api url on the docker block, as 0.0.0.0:4200/api, and started Orion on 0.0.0.0:4200 - also set network mode on the docker block to "bridged" (saw it somewhere on the internet). No love. Any ideas?

Hey folks -- I feel like I've missed something obvious, but how do I configure Prefect 2.x to run successfully behind a reverse proxy? I see references to

PREFECT_SERVER__BASE_URL

but that seems to be for v1. I'm using Traefik to reverse proxy https traffic to Prefect. The UI loads, but I see this error displayed: "Can't connect to Orion API at http://0.0.0.0:4200/api"

Hello, Like we used to pass memory_limit and memory_request in run_config in Prefect1, is there something similar we can do in Prefect2 while doing cli deployment using overrides or something? If so, can someone please help with sample?

[prefect 2.0] running into some weirdness with work queue concurrency. I have a work queue with concurrency set to 1, and at some point this morning, jobs on this queue were not getting picked up by agents. If i set concurrency to 2, then the new runs start to get picked up. I don't see any jobs in

pending

or

running

, any ideas on how i figure out what is blocking this queue from starting new runs?

TypeError: Type is not JSON serializable: numpy.float64

This error is thrown on a line in a flow that simply passes 3 dataframes into a function call. Like nothing within the code, it's the line that calls the function. Does anyone know what this is?

I'm submitting ~1000 tasks with ConcurrentTaskRunner but they all tasks stay stuck in status "submitted". I expect the tasks to be executed right away but this is not the case. Each tasks is responsible for doing a web request to download an image. Why is ConcurrentTaskRunner not able to handle this?

Hi Prefect, It looks like the “notification” panel on the left menu just disappeared, is that a known bug ?

Hi, Getting an error on a simple task, that is not an error with the actual task but appears to be something with the Prefect 2 internal logging. (as far as I an see so far) The task writes an reads info from a postgres DB, but close the connections after doing so. This issue did not happen when the task was running in Prefect 1 Any thoughts?

07:51:37.133 | INFO    | Flow run 'ultra-honeybee' - Finished in state Completed('All states completed.')
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/site-packages/IPython/core/history.py", line 780, in writeout_cache
    self._writeout_input_cache(conn)
  File "/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/site-packages/IPython/core/history.py", line 763, in _writeout_input_cache
    conn.execute("INSERT INTO history VALUES (?, ?, ?, ?)",
sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 123145498836992 and this is thread id 140704730421440.

Hey community! How do I check using Orion/Prefect-API if a flow has finished or not?

Hi , I wanted to accept a parameter to a master

prefect

flow and set the reference tasks parameters based on that.

with Flow(
    "Master",
    schedule=schedule,
    run_config=master_run_config,
    terminal_state_handler=_utils.final_state_handler,
) as flow:
    logger = prefect.utilities.logging.get_logger("master")

    bot_ids = Parameter("Bot Ids", required=False, default="")
    bots = bot_ids.run() if bot_ids else ""
    bots = [x for x in bots.split(",") if x]
    bots = [x.strip() for x in bots]

    
    all_b = ['A','B','C','D']
    unknown_bots = [b for b in bots if b and b not in all_b]

    if len(unknown_bots) > 0:
       #TODO  raise exception

    #TODO set reference tasks

From the above code snippet, I’m expecting a parameter

Bot Ids

and checking if the supplied

Bot Ids

are valid or not. Here I’m facing two issues: •

Bot Ids

parameter is not available in the prefect UI. I tried passing

Bot ids

to one of the reference tasks (Though I’m not using the parameter inside the task), and the parameter started appearing in the prefect Run Flow UI. is there any way to make the parameter input available in prefect Ui without having to pass the parameter to any of the reference tasks • I tried passing values for

Bot Ids

from the prefect UI and noticed that it was not getting captured in the

master

flow. I was always getting

bot_ids

as empty • I tried changing the default value of the parameter , and noticed that the flow was working as expected.

bot_ids = Parameter("Bot Ids", required=False, default="steps")

Why is that the default value works in this case, but not the supplied values?

Is it possible to log the function name and line number? (I know I can do this with the default logger, but is this possible with prefect logger?) also, can this be done with @flow(log_prints=True)?

Silly question, where does Prefect fit in with Python Celery and Message Queues / Brokers (e.g. RabbitMQ)? It seems like some of the tasks, such as restarting (or retrying) a long running asynchronous job can be implemented in either of the above solutions. I was wondering what would be the difference and the tradeoffs to using each of the above solutions.

Naive Question, How does the prefect work queues process the job, like i have a inferencing pipeline set with 1 work-queue, so if inferencing got triggered from let say 10 source then how will the queue handle it, is it do round-robin, sequentially or something else ?

Have anyone been able to log pyspark stderr to prefect log? Trying to track down some issues and it turns out that pyspark write to stderr for Info. I am still on prefect v1.

hey folks... me and a teammate are experiencing a strange issue where logs are not being displayed from within running functions (we've tried warning and error). has anyone else observed this behavior? this is in prefect 2 specifically, and the pattern we're using is typically a task which calls a different function

Prefect 2.0. Is there a way to have a dropdown in the cloud interface for parameters instead of just a text box?

Is there a way to ask "map" to create batches instead of one task per value? For example if I have 1M events, I feel that seeing 1M tasks in the UI might make things bad. Maybe I am wrong on that From that

@task
def A() -> List[int]:
    return [1, 2, 3]

@task
def B(y: int) -> int:
    return y + 100

with Flow('flat map') as f:
    a = A() # [1, 2, 3]
    b = B.map(x=a) # [101, 102, 103]

to

@task
def A() -> List[int]:
    return [1, 2, 3]

@task
def B(values: List[int]) -> List[int]:
    return [y + 100 for y in values]

with Flow('flat map') as f:
    a = A() # [1, 2, 3]
    b = B.map(x=a, batches=2) # [101, 102, 103]

Quick question, Using prefect 2.0. If I run a flow locally, it is possible to access the flow_run_id of that run?

With Pv2, when you alter a block, do you have to restart the agent? Rebuild the deployment using the block?

I have

export PREFECT__LOGGING__LEVEL=DEBUG

def main(): 
    logger = get_run_logger()
    logger.warning("warning")
    logger.debug("debug")
    <http://logger.info|logger.info>("info")
    logger.critical("critical")

yet the debug isn’t printing:

14:49:20.921 | WARNING | Flow run 'wonderful-hamster' - warning
14:49:20.921 | INFO    | Flow run 'wonderful-hamster' - info
14:49:20.922 | CRITICAL | Flow run 'wonderful-hamster' - critical

Is there a way to customize the Orion logging UI? right now it only shows the level (info), message, and time. But I’d like it to show everything I’m putting in my log.

Initializing flow runs with persisted state/results programmatically in Prefect 2.

Hi everyone I'm trying to get a sample flow run using Prefect UI, S3 as storage and Kubernetes Job as Infra. Unfortunately I get the

ModuleNotFoundError: No module named 's3fs'

when running the flow. • I am not using a custom image for the agent or the the job and prefer to keep using official images. • I have specified the env for the kubernetes job in the Prefect UI but it does not make a different. • I have also tried adding the

pip install s3fs

command to the agent's deployment manifest but that doesn't make a difference either. Any ideas?

Hi , I wanted to accept a parameter to a master

prefect

flow and set the reference tasks parameters based on that.

with Flow(
    "Master",
    schedule=schedule,
    run_config=master_run_config,
    terminal_state_handler=_utils.final_state_handler,
) as flow:
    logger = prefect.utilities.logging.get_logger("master")

    bot_ids = Parameter("Bot Ids", required=False, default="")
    bots = bot_ids.run() if bot_ids else ""
    bots = [x for x in bots.split(",") if x]
    bots = [x.strip() for x in bots]

    
    all_b = ['A','B','C','D']
    unknown_bots = [b for b in bots if b and b not in all_b]

    if len(unknown_bots) > 0:
       #TODO  raise exception

    #TODO set reference tasks

From the above code snippet, I’m expecting a parameter

Bot Ids

and checking if the supplied

Bot Ids

are valid or not. Here I’m facing two issues: •

Bot Ids

parameter is not available in the prefect UI. I tried passing

Bot ids

to one of the reference tasks (Though I’m not using the parameter inside the task), and the parameter started appearing in the prefect Run Flow UI. is there any way to make the parameter input available in prefect Ui without having to pass the parameter to any of the reference tasks • I tried passing values for

Bot Ids

from the prefect UI and noticed that it was not getting captured in the

master

flow. I was always getting

bot_ids

as empty • I tried changing the default value of the parameter , and noticed that the flow was working as expected.

bot_ids = Parameter("Bot Ids", required=False, default="steps")

Why is that the default value works in this case, but not the supplied values?

Prefect 2 migration Hi all, given the recent announcement regarding the Prefect 1 freeze I am looking for any insight on difficulties when migrating. The announcement states that most Prefect 1 flows are compatible with Prefect 2 - which types of flows/features are not? What kind of issues are common when migrating? Cheers! 🙂

Hi, I have a question regarding subflows and deployment: say in my flow I want to launch a subflow, but this runs on a completely different architecture than the original flow - so the most appropriate thing would be to use a different deployment. I see I can launch deployment using the API, but this then won't really be connected to the original flow?

Where does prefect decide which packages to install?

Hey prefect guys .... Does anyone knows how to prevent a pod/job to be moved to another node-pool in k8s ? I am facing this problem ... I already tried to run with podDisruptionBudget but not worked... I am trying to create AntiAffinity and also it is not working ... Thank you guys.