Hey all! Running into an issue with our prefect deployments that I am wondering if anyone else has seen or has any thoughts on. We have a large quantity of flows that are running/queued, but each flow itself should not take very long. For some reason a large number of these flows are entering a running state and then never exit that state. The associated Job deployment for those flows gets cleaned up, but the flow itself still exists as running and is blocking the queue

@Marvin: the docker run as task in prefect could not write out output to local file system. Could you please give me an example to execute prefect flow using local built docker image, with a task read input and write output to the local file system?

@Marvin Using prefect 3. I set up a k8s pod that serves a flow. How can I limit the n. of flow runs per pod? I want at most 1 flow run per pod (I want to scale it with more pod replicas).

@Marvin how can I serve multiple deployments of the same flow using the .serve method?

@Marvin How can you show dependency arrows in the flow in prefect v2.X?

@Marvin I am noticing a lot of sqlite database locks which is adversely affecting the visual state and reporting of flows, subflows and tasks in the UI. For example. I have a flow with multiple concurrent tasks that run. They do appear to be running however the UI shows that there are 0 tasks running and there is no activity on the graph. How can I add a postgres server to my prefect instance and migrate from sqlite to the postgres instance while keeping the blocks, variables, deployments and other details in place?

Hello, I have this flow which uploads data to GCS:

from pathlib import Path
from prefect import flow
from prefect_gcp import GcpCredentials, GcsBucket
import pandas as pd

@flow
def upload_to_gcs():
    name = 'ready_df.csv'
    file_path = Path(name)
    df = pd.read_csv(file_path)
    gcp_credentials = GcpCredentials.load("...", validate=False)
    gcs_bucket = GcsBucket(
        bucket="...",
        gcp_credentials=gcp_credentials
    )

    gcs_bucket.upload_from_dataframe(df=df, to_path=name)
    
    return None

if __name__ == "__main__":
    upload_to_gcs.serve(name="...")

To run this I install prefect-gcp, and with it the prefect version that comes is 2.19.7. My project is using 3.x. So if I upgrade to 3.x in the terminal where I just installed prefect-gcp, when running python this_flow.py I get:

from prefect.infrastructure import Infrastructure, InfrastructureResult
ImportError: cannot import name 'Infrastructure' from 'prefect.infrastructure'

and I cannot .serve() it while having version 2.19.7 because I guess .serve was not available then and I get:

prefect.deployments.runner.DeploymentApplyError: Error while applying deployment: Client error '422 Unprocessable Entity' for url '<http://127.0.0.1:4200/api/deployments/>'

Is this a known issue with prefect-gcp and prefect 3.x? Thank you

@Marvin In prefect 3.0 how can I export blocks and import them into a different prefect instance

@Marvin What does it mean when my deployments have “not ready” as the status?

@Marvin, is configuring the logger to print json as simple as setting an env var? Looking at the logging.formatters here: https://docs.prefect.io/latest/api-ref/prefect/logging/formatters/#prefect.logging.formatters.JsonFormatter and this config file: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/logging/logging.yml