Hi there! Ever since we have scheduled a flow in a higher frequency, we see a lot of "*late*" flow runs in the prefect UI. What does this status mean? It feels like late flow runs are not automatically re-running. Also, it feels like all these late flow runs are "blocking" new flow runs from running (without the concurrency limit reached). Sometimes running a deployment in the prefect UI immediately results in a late state flow run

Hello 🙋 let's say i have a mapped task

X

that spawn several "children" tasks

Y1

...

Yn

let's say some of them are stuck on

Running

as a finished state - but i wanna guarantee the next phase in the flow executes correctly and ignores the few "hung" tasks -- what's the (default) behavior for if i set those tasks (e.g.

Y3

,

Y17

and

Y91

) to

Skipped

? would the next task that depends on them still get executed (even if it has the default

all_successful

trigger?) the reason i'm asking about

Skipped

is because i wanna avoid a

None

response flowing downstream from these tasks

Hey Prefect, I have tried disabling heartbeat using instructions at https://docs.prefect.io/orchestration/concepts/services.html#heartbeat-configuration, however they dont seem to work for Prefect Server. What’s the

config.toml

setting for disabling both heartbeat and lazarus for prefect flows on Prefect Server?

For a private db in AWS can I connect to it using MysqlFetch in a function without

@task

decorator ? I get

pymysql.err.OperationalError: (2003, "Can't connect to MySQL server on '<http://xyz-mysql8-cluster.cluster-fa43asdfd.us-west-2.rds.amazonaws.com|xyz-mysql8-cluster.cluster-fa43asdfd.us-west-2.rds.amazonaws.com>' (timed out)")

every time. Any suggestions on how to approach in solving this?

hi all We are using an Automation for a DAG to create a new flow-run after a flow-run has “Finished”, We are facing an issue where multiple flow-runs are created at the same time

Guys do prefect have priority flows? Example i have 4 flows: a, b, c, d and i want to prioritize a and c to agent run first is this possible?

I'm having trouble getting prefect agent logs for the debug level to show up. I've set

PREFECT__LOGGING__LEVEL=DEBUG

as an environment variable, confirmed it's in the environment, and my agent logs are not showing anything but INFO. Is there anything else that I should need to set?

Hello 🙂 I'm trying to build a Graphql query to compute the number of running tasks for a particular task. I tried with the

task_run_state_aggregate

but I don't know how to filter on a specific task 😕 If anyone can help 🙏

I'm running into a case where the prefect cloud UI shows jobs as submitted. I have queried the graphql api by the flow id to confirm it is submitted (i presume that's where the ui gets it's info from anyway). The agent however can finds no ready flows. I am running with debug mode to see that it's polling and never finding anything:

[2022-07-14 17:49:44,859] DEBUG - agent | Querying for ready flow runs...
DEBUG:agent:Querying for ready flow runs...
[2022-07-14 17:49:44,959] DEBUG - agent | No ready flow runs found.
DEBUG:agent:No ready flow runs found.
DEBUG:agent:Sleeping flow run poller for 10.0 seconds...
[2022-07-14 17:49:44,960] DEBUG - agent | Sleeping flow run poller for 10.0 seconds...
[2022-07-14 17:49:54,960] DEBUG - agent | Querying for ready flow runs...
DEBUG:agent:Querying for ready flow runs...
[2022-07-14 17:49:55,046] DEBUG - agent | No ready flow runs found.
DEBUG:agent:No ready flow runs found.
[2022-07-14 17:49:55,046] DEBUG - agent | Sleeping flow run poller for 10.0 seconds...
DEBUG:agent:Sleeping flow run poller for 10.0 seconds...

Anyone have an suggestions on how to continue to debug?

Does anyone have any tips working with a random number generator in prefect. I want my tasks to be able to recreate the result and also make sure they don’t get the same value when concurrency is involved. One of the things I though about is generating a set of random numbers based on a seed and passing them along to specific tasks. That way concurrency isn’t an issue. Just feel it’s not a very elegant solution. I though about putting the RNG in the context but then concurrency might be an issue

Does anyone know on top of their head why a script with

flow.run()

will fail once I register it to the UI? I'm current using the local server

prefect backend server
prefect server start
prefect agent local start

the error im getting is

Failed to load and execute flow run: FlowStorageError('An error occurred while unpickling the flow:\n ModuleNotFoundError("No module named \'utils\'")\nThis may be due to a missing Python module in your current environment. Please ensure you have all required flow dependencies installed.')

Hi I'm trying to prepare a parametrised flow for Prefect v1. It has almost the same code as https://github.com/anna-geller/packaging-prefect-flows/blob/master/flows_task_library/s3_kubernetes_run_RunNamespacedJob_and_get_logs.py , but I only changed some functions to make them ready to work with parameters. I cannot get why does not it work now: function

CreateAndRunJob

marked as succeeded but prefect does not wait for the end of the execution and then function

get_our_pod_name

fails because it cannot find Job which is not yet created. I though to use threading or queue from python here, but I think it's not the right way with Prefect. Could you please advice here? I believe I do something wrong here.

Hi! I'm trying to use the Orion python API

prefect.packaging.docker.DockerPackager

class and have it push to Google Container Registry. It looks like the current

package()

method is unable to push to GCR because the

prefect.docker.push_image

here ignores any path provided in the registry url. So it looks to me like if I provide a

registry_url

like

<https://gcr.io/my-project-name>

(where

my-project-name

is a GCP project) for a flow named

my-flow

rather than pushing an image to

<http://gcr.io/my-project-name/my-flow|gcr.io/my-project-name/my-flow>

the current implementation will try to push to

<http://gcr.io/my-flow|gcr.io/my-flow>

(which fails because I don't have a project named

my-flow

). Wondering if I'm missing something here or maybe this should go into an issue?

hello everyone, i have been trying to set up the prefect server. When I run prefect server start, I encountered this error. I have already installed docker-compose.

Are there any suggestions for where to put code for instrumentation? I'd like to wrap my flows and tasks. I've found two spots that seem ok:

def patch_task():
    method = getattr(CloudTaskRunner, "run")

    def patched(self, *args, **kwargs):
        name = self.task.name
        with tracer.trace(f"task", resource=name):
            result = method(self, *args, **kwargs)
        tracer.flush()
        return result

    setattr(CloudTaskRunner, "run", patched)


def patch_flow():
    method = getattr(CloudFlowRunner, "run")

    def patched(self, *args, **kwargs):
        name = self.flow.name
        with tracer.trace(f"flow", resource=name):
            result = method(self, *args, **kwargs)
        tracer.flush()
        return result

    setattr(CloudFlowRunner, "run", patched)

Hi everyone, a question about Prefect 1. I am trying to run a subflow using a task. My code:

@task(log_stdout=True)
def create_subflows(list_of_elements):
   list_param= []
   for element in list_of_elements:
        list_param.append(
            {
                "pr": element,
            }
        )
   mapped_flows = create_flow_run.map(
        flow_name=unmapped("Subflow"),
        parameters=list_param,
    )

with Flow("Parent Flow") as flow:
   list_of_elements = [1,2,3]
   create_subflows(list_of_elements)

I am getting this error: `ValueError: Could not infer an active Flow context while creating edge to <Task: create_flow_run>. This often means you called a task outside a with Flow(...) block. If you're trying to run this task outside of a Flow context, you need to call

create_flow_run.run(...)

Just an interesting observation. When using clocks.RRuleClock for creating built-in scheduler and setting start time to 2000-01-01, when deployed via "prefect register" it takes 10 minutes for deployment. When doing the same with 2022-01-01, it takes a minute. Wonder what is happening under the hood 🙂

Hi all~ I want to know that how to change the prefect's default server database! (using prefect 1.0) There's method "Running Prefect with a pre-existing postgres database" in prefect docs. But can i change the

server.database.connection_url

or

server.database.host

to connecting information for MySQL database, not postgres?

Hi everybody, to we have exemples of Prefect Orion used to manage Airbyte and dbt tasks ? I have this simple need and I'm wondering if I should use Prefect 1 or 2.

In Prefect 1, I can’t seem to see any logs in the UI when using

LocalDaskExecutor(scheduler="processes")

while with

threads

logs can be seen in UI as normal, any ideas?

Hi, I'm wondering whether the following behaviour is actually wanted (and intended), because it seems weird to me (Prefect 2.0b8, server): Assume I have a file with a flow and a deployment, when executing

prefect deployment create file.py

the deployment gets created. Now, when I modify the flow a little (flow name stays the same), change the deployment name and then re-execute the above command, a new deployment is created. This deployment points at the same flow object (flow_id is the same for both). However, both deployments execute different code, 'different versions of the same flow' if you will, although this 'version number' is not saved anywhere (I think). This all is due to the fact that the location of the flow code (flow_data) is saved with the deployment, and not with the flow, which seems a little counterintuitive for me. If I see 2 flow runs in the UI that executed the same flow, I would expect them to have executed the same code.

I need to scale out a Flow such that I significantly increase the parallelization. I’m already mapping tasks and adding memory to machines to be able to handle more threads. Supposing I didn’t want to keep scaling a single machine, is there a way to map tasks out to multiple machines? What do I need to be thinking about?

is there a way to migrate your project from pefect v1 to prefect orion ?

This is more a roadmap/feature question about prefect 2.0. Will the notification system have a more generic option like webhook? If i could setup a webhook this could open up a lot of possibilities for dealing with certain types of failure conditions.

from prefect.storage import GitHub,GitLab, S3, Webhook

from prefect.backend.artifacts import create_link_artifact
import prefect.engine.cache_validators


@task( task_run_name="mviz_task_{md5}", max_retries=0, cache_for=datetime.timedelta(hours=1),
      )
def test( md5,   ):
    print('do test', md5 )
    

with Flow("epl" ) as flow:
    test( '123' )
    test( '123' )
flow.run()

hi, i try to use cache in a single flow run, but i got the belllow output which shows that cache not work. it seems that cache will not work durning same flow run or there are some mistakes in my code ?

[2022-07-15 12:18:00+0000] INFO - prefect.FlowRunner | Beginning Flow run for 'epl'
[2022-07-15 12:18:00+0000] INFO - prefect.TaskRunner | Task 'test': Starting task run...
[2022-07-15 12:18:00+0000] WARNING - prefect.TaskRunner | Task 'test': Can't use cache because it is now invalid
do test 123
[2022-07-15 12:18:00+0000] INFO - prefect.TaskRunner | Task 'test': Finished task run for task with final state: 'Cached'
[2022-07-15 12:18:00+0000] INFO - prefect.TaskRunner | Task 'test': Starting task run...
[2022-07-15 12:18:00+0000] WARNING - prefect.TaskRunner | Task 'test': Can't use cache because it is now invalid
do test 123
[2022-07-15 12:18:00+0000] INFO - prefect.TaskRunner | Task 'test': Finished task run for task with final state: 'Cached'
[2022-07-15 12:18:00+0000] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded

Good morning, I've got this issue. I am using LocalDaskExecuter() and Git Storage. When I run the flow in the cloud, dask breaks down a task into multiple ones, with new labels. But the flow in Git storage doesn't recognize the new tasks, which gives me a mismatch error between the flow version stored in prefect backend and the flow version in storage. How do I get around that?

Are labels applied in the GUI expected to apply to scheduled flow runs?

Regarding Parameters, is there any way to get the result during a flow? I'm trying to put various parameters into a Pydantic BaseModel that we're using as a config object for a task, but the BaseModel is failing the validation because the type is Parameter instead of str, for example.

I'm running into an issue where "ghost" tasks/flows are taking up my concurrency limits. I'm using prefect cloud 1.0 and using either kubernetes or local agents for my flows. Some issues I'm running into: • Some of my task concurrency limits have been used up for many weeks in a row, even when I have no running flows at all - and especially not any that have that task tag. I have tried the following but the slots still seem to be used up: ◦ I have stopped all runs in progress using the UI ◦ For my flows deployed on k8s, I have removed all jobs that have been running for >x days (these were "ghost" jobs too as there was no flow that was running that long) ◦ I have restarted my agent • Some cancelled flows are still showing up in the Running tab. They're greyed out and have the "cancelling..." written below them for quite a few days. My question: • How can I identify why my task concurrency slots are being used and how can I clean them up?

Is it possible to create a new service account key from the prefect CLI or the API?