Hi All. I have an issue with failed flows not deleting their pods in kubernetes. If I run a flow, and then start another run of the same flow, the second (or any subsequent) run will fail with kubernetes throwing a 409, but the pod it was running on doesn’t get deleted and remains in running state - any suggestions how to get around this?

He all, Quick question: If I have several tasks inside a prefect flow, and I want the downstream tasks to wait for some upstream tasks (which are mapped), what can I do? These are not sequential flows ... to be able to use the set.upstream and wait for flow. Any help is appreciated.

Is

databricks

currently supported as a connection/profile type when using prefect-dbt cli commands (like

trigger_dbt_cli_command

?

RuntimeError: Command failed with exit code 1:
    Could not find adapter type databricks!

Is there any documentation on how to generate an API key for a self hosted graphql server?

Does anyone recommend any particular consulting groups/agencies around data engineering/data warehouses?

Is it possible to specify separate

KubernetesJob

infra blocks in a

Deployment

? e.g.

flow a
  task b (uses KubernetesJob with docker image foo)
  task c (uses KubernetesJob with docker image bar)

It kind of seems like I'm limited to to having one

KubernetesJob

per

Deployment

?

In Prefect 2.0, is it possible to schedule an adhoc flow run at some time in the future?

Hello, I am new to using Prefect. I have a task defined as below. I noticed when it fails and retries, the input value got lost. original run

my_task(run_date="20221004")

retrying run:

my_task(run_date=None)

Any idea what I might have done wrong, or any workaround?

@task(retry_delay=timedelta(minutes=30), max_retries=10)  # wait for up to 5 hours
def my_task(run_date: str):

Hi, is there a nice pod spec for setting up an ephemeral dask cluster using the latest prefect/prefect_dask versions?

Hi All, I'm having trouble trying to dynamically build flows from a list of tasks. For example, I have tasks 1, 2, and 3 that will all modify the same dataframe. However, they might need to run in different orders or not all depending on user input. Ideally id like to build a list of these tasks to run later. For instance if I want to run task 1, 3, and 2 in that order (with their associated parameters) I'd like to call a task runner with a list of tuples like: [(Task_1, Config_1), (Task_3, Config_3), (Task_2, Config_2)]. I'd appreciate if anyone could point me in the right direction. Thanks!

hi folks, first of all, wanna thank you for creating such a great tool! I recently realized that besides the workflow orchestration, the block is a really convenient feature for me to store sensitive data. I am wondering, if it possible to use the Prefect block as a standalone module? e,g. using

Secret.load("github-access-token").get()

in normal python scripts?

Hi all, I’m running a Prefect (1.0) agent on a VM in Azure. After around a week it gets stuck with this error. Once I restart the process it works perfectly fine for another week until it gets stuck again. Anyone have a clue what is happening? Thanks!

Hi, I have a flow run whose all of the child tasks were completed but the flow itself is still flagged as Running. what's the possible cause for this? How can I flag this run as finished?

Hello all, is there a way to pass

timeout_seconds

of the flow as parameter to customize it when trigger a deployment of that flow manually from UI?

@flow(timeout_seconds = 15)
def log_flow(name: str):
    log_task(name)

As timeout_seconds is in the flow decorator, it won't be shown as a parameter of the flow run

the options for cli

prefect agent

are:

╭─ Options ──────────────────────────────────────────────────────────────────────
│ --work-queue    -q      TEXT  One or more work queue names for the agent to pull from. [default: None]                                                                                                                                           │
│ --hide-welcome                                                                                                                                                                                                                                   │
│ --api                   TEXT  [default: (from PREFECT_API_URL)]                                                                                                                                                                                  │
│ --tag           -t      TEXT  DEPRECATED: One or more optional tags that will be used to create a work queue [default: None]                                                                                                                     │
│ --help                        Show this message and exit.                                                                                                                                                                                        │
╰─────────────────────────────────────────────────────────────────────────────────

What I would like to do is launch an agent once, grab the flow_runs that are sitting in the queue (a single call), once those are completed, the agent would close itself. Is there any kind of option for that?

Is there a limit on data size for params for a subflow? I have a flow which passes two dataframes to a subflow, and one of the dataframes causes an error to be thrown:

TypeError: Dict key must be str

but the other doesn't. The one causing the error does have ~1 million rows though.

👋 My flow is a part of a python package and uses utility functions which are also a part of the package (e.g.

projname/flows/myflow.py

imports

projname.utils.utilfunc

). This flow is scheduled from a deployment and uses

Process

as its infrastructure block. I would like to ensure that the latest version of

projname

is installed (to guarantee that

utilfunc

is up-to-date), so I can't install the package when setting up agent environment. I think I can install the package before running the flow if I set

Process.command

to something like

["pip", "install", ".", "&&", "python3", "-m", "prefect.engine"]

, but maybe there's a better way to go about it?

@Will Raphaelson let me in

Running a very basic flow in Prefect Cloud with prefect agent deployed in EKS. In our logging infra, I’m getting

ERROR prefect.infrastructure.process - Process 'turquoise-woodlouse' exited with status code: -9

No more messaging beyond that. In Prefect Cloud UI, the flow runs indefinitely and doesn’t show this log message (level set to DEBUG). Anyone familiar with exit code -9? Can’t find any docs.

Noob Question - Thought I'd ask this here rather than on Github Issues because I'm fairly certain this is user error on my part. Trying to setup the dev environment but when running pytest I get the following errors. Any thoughts on what I'm doing wrong?

ImportError while loading conftest '/home/gary/prefect/tests/conftest.py'.
tests/conftest.py:32: in <module>
    import prefect
src/prefect/__init__.py:23: in <module>
    from prefect.flows import flow, Flow
src/prefect/flows.py:36: in <module>
    from prefect.context import PrefectObjectRegistry, registry_from_script
src/prefect/context.py:34: in <module>
    from prefect.client import OrionClient
src/prefect/client.py:55: in <module>
    from prefect.orion.api.server import ORION_API_VERSION, create_app
src/prefect/orion/api/__init__.py:1: in <module>
    from . import (
src/prefect/orion/api/block_documents.py:19: in <module>
    from prefect.orion import models, schemas
src/prefect/orion/models/__init__.py:1: in <module>
    from . import (
src/prefect/orion/models/block_schemas.py:12: in <module>
    from prefect.blocks.core import Block
src/prefect/blocks/__init__.py:3: in <module>
    import prefect.blocks.notifications
src/prefect/blocks/notifications.py:26: in <module>
    class SlackWebhook(NotificationBlock):
pydantic/main.py:283: in pydantic.main.ModelMetaclass.__new__
    ???
../anaconda3/lib/python3.9/abc.py:106: in __new__
    cls = super().__new__(mcls, name, bases, namespace, **kwargs)
src/prefect/utilities/dispatch.py:99: in _register_subclass_of_base_type
    register_type(cls)
src/prefect/utilities/dispatch.py:154: in register_type
    key = get_dispatch_key(cls)
src/prefect/utilities/dispatch.py:76: in get_dispatch_key
    dispatch_key = dispatch_key()
src/prefect/blocks/core.py:127: in __dispatch_key__
    return block_schema_to_key(cls._to_block_schema())
src/prefect/blocks/core.py:263: in _to_block_schema
    return BlockSchema(
pydantic/main.py:340: in pydantic.main.BaseModel.__init__
    ???
pydantic/main.py:1076: in pydantic.main.validate_model
    ???
pydantic/fields.py:884: in pydantic.fields.ModelField.validate
    ???
pydantic/fields.py:1101: in pydantic.fields.ModelField._validate_singleton
    ???
pydantic/fields.py:1148: in pydantic.fields.ModelField._apply_validators
    ???
pydantic/class_validators.py:318: in pydantic.class_validators._generic_validator_basic.lambda13
    ???
pydantic/main.py:690: in pydantic.main.BaseModel.validate
    ???
E   DeprecationWarning: `copy_on_model_validation` should be a string: 'deep', 'shallow' or 'none'

Is there a way to monitor for lack of logs in a prefect task (or better, in a flow)?

Hey all, just recently set up Orion in our EKS cluster to pilot switching over to RayTaskRunner. We were experiencing some pretty untenable memory bloat with high numbers of tasks with Dask (25K+). My question about Ray: does the local cluster capability feature the same worker adaption that Dask does? In my testing it seems to function basically the same as a concurrent executor (single pod, local scheduling). Am I missing something here? Or is this a case where we need to deploy a local Ray cluster to get that use case?

Hi all! Are there any code examples of using the python client to create a flow run using

create_flow_run()

in prefect 2? I’m trying to see how the

flow

parameter should be setup

Hi All, Whats the best way to get information such as flow id and task id programmatically during execution?

Hi everyone! I'm sure this has been answered but I’m struggling to find in the documentation. I want to schedule a flow, but I only want to have that flow run once another flow in my Prefect Cloud instance has completed successfully. Can someone point me in the right direction? I’m using Prefect 1. Thanks!

Hi everyone! I'm using Prefect V1 and I'm having some issues with the timeout configuration. Our task is kicking off

ShellTask

which is written in Typescript. As part of the typescript script we call python script. The prefect timeout works if it's reached while typescript is executing, but the timeout doesn't work if the process is stuck inside the python script. Has anyone ran into the same issue before in the past? Any advice is appreciated, thank you in advance!

Is there an example of using a private docker container registry on Prefect V2?

Hello everyone, I’m having trouble getting the

prefect agent

to execute within my Django environment (using Django ORM in the flow). I finally got the deployment created by adding my project to the sys path before running the Django setup. That works for running the flow manually, but now I’m getting another error when the Agent tries to run the flow. Full error in the thread message. So my idea at this point is to create a Django Management command (which executes in the Django environment) to run the agent through Python code. I have an example of doing this in Prefect v1, but I’m using Prefect v2 and haven’t see where this is clearly documented.

LocalAgent

no longer seems to exist in Prefect v2 to run an agent through code. Any ideas of other ways to run this or how to run the agent from within Python would be super helpful and wholly appreciated!

hey all - I have a prefect orion server running in docker. I’m trying to run hello world flows outside of the orion docker, just a local flow but I want the flow to be connected to the orion server. From what I found I have to set the config variable for prefect with prefect config set PREFECT_API_URL = “” but I can’t figure out what the url is suppose to be.

Hey guys, what’s the best workaround of accessing the code in a private git repo?