Is there a setting for registering a workflow at a certain interval (say every 1 hr) but having the workflow start instantly then wait 1 hr for the next run?

Hi, today I'm getting errors about Prefect not finding my secrets: KeyError: 'The secret xxx was not found. Please ensure that it was set correctly in your tenant: https://docs.prefect.io/orchestration/concepts/secrets.html'. This happens during a flow where I use the same secret multiple times to access a data source, after a while I get this error. But it must have found the secret before, since the data source was accessed earlier in the flow. Anyone else getting the same errors?

Hey everyone, we have a flow that creates another flow and reacts to its results using

create_flow_run

and

get_task_run_result

. We unit tested the parent flow using mocks to simulate the possible results of the child flow. However, it would be great to have a basic integration test where we ideally only mock the outside dependencies like sql and api calls. Do you have any ideas how to achieve that?

Hi, I need some help forcing a linear dependency between my tasks:

execution_id = task_run_inference()
task_decision_process(execution_id, config, upstream_tasks=[task_run_inference])

    task_append_to_m2m(execution_id, config, upstream_tasks=[task_decision_process])

    task_append_to_m2r(execution_id, config, upstream_tasks=[task_decision_process])

    task_run_inference_cleanup(execution_id, config, upstream_tasks=[task_append_to_m2r, task_append_to_m2m])

This is returning a weird dependency plot where task_run_inference is pointing to task_append_to_m2r. Do I need to explicitly set all tasks as upstream? wont this create a dependency between them?

Hello, I do need some information for passing "endpoint_url" to prefect.tasks.aws.s3.S3List.run to change default endpoint URL. Is it possible in PREFECT? Like `aws s3 ls --endpoint-url https://s3.xxx.xxx.xxxx/`in aws cli. Thx a lot

Hello everyone, I am new to the prefect and find it interesting. One question I have is, how do I push different versions of workflow code? Whenever I push a new version of code, prefect cloud does not pick it up. I have to create a new flow to reflect the change. 1. Is there a way to update the current version of code? 2. How to push a new version of the code? I am using prefect cloud

Hello, hopefully an easy question. What backing cloud provider is Prefect Cloud using? I need to be able to answer this for bureaucratic reasons.

I’m having some difficult deploying the prefect agent into Kubernetes using a namespace and service account. When trying to trigger a flow I get the following error. It looks like it’s trying to use the

default

service account in the

prefect

namespace I’m attempting to use:

system:serviceaccount:prefect:default

(403)
Reason: Forbidden
HTTP response headers: HTTPHeaderDict({'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Content-Type-Options': 'nosniff', 'X-Kubernetes-Pf-Flowschema-Uid': '0cb3f7e4-3156-4e9e-b025-5c9deb274813', 'X-Kubernetes-Pf-Prioritylevel-Uid': '96071337-eda2-47ce-9026-95ed0ab85b02', 'Date': 'Fri, 03 Sep 2021 19:49:12 GMT', 'Content-Length': '311'})
HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"jobs.batch is forbidden: User \"system:serviceaccount:prefect:default\" cannot create resource \"jobs\" in API group \"batch\" in the namespace \"prefect\"","reason":"Forbidden","details":{"group":"batch","kind":"jobs"},"code":403}

I specified SERVICE_ACCOUNT_NAME in the kubernetes manifest but it doesn’t seem to be honoring it. Am I perhaps missing another configuration?

quick question - when adding a label to a flow, what is supposed to happen to future scheduled flows? Should they get the labeled applied automatically? I had thought they would, but I had to turn off the schedule and then turn it back on for new flow runs to generate with the new label

Has anyone noticed any degraded service on prefect cloud. It seems that my flows keep exiting before completion (no error message on cloud)

Does Prefect have something analogous to the Airflow Operator for starting, monitoring, and terminating AWS EMR clusters?

Has anyone experimented with rigging prefect flows which execute jupyter notebooks? We have folks who author scrapes in a no-code tool, and then they may need to set up scheduled postprocessing of incremental ingested data. Anyone doing something similar?

@channel Experiencing a weird issue with Prefect Cloud. We have twice experienced flows that have stopped scheduling themselves ~1.5 hours after the scheduler is enabled (see image). This flow was deployed yesterday afternoon and is scheduled to run every 10 minutes. The first instance of schedule “running out” happened at 2:40 PM (PST). Disabling and renabling the flow schedule allowed more flow runs to be schedule. This happened again at 4:40 PM PST. Tagging @nicholas and also going to follow up with sales@prefect.io. Hoping to get this resolved soon so my devs and I can avoid flipping switches every 1.5 hours and enjoy our long weekend 😉

Hi Everyone, I am using Webhook Storage to register flows on our onprem prefect server, and the flows are configured to used Kubernetes Run .. .I am trying to understand who does the actual GET API Call to the webhook to download the file (Agent Pod, Job Pod) and when does that happen? Does it happen when the flow is about to run or when the flow is being registered?

I'm wondering if anyone has a custom job_template YAML for KubernetesConfig that they used to assign some environment variables. I can find it in the documentation for the legacy components but not for the current one. Thanks

Using prefect, is it possible to implement tasks in different programming languages, we already have our business login in Nodejs and I couldn't find any docs related to this

Hi ! I'm trying to register a flow from a flask API. This flow register successfully, but i always get an error in the prefect logs (c.f. picture). Do you have any idea ? Tell me if you need more informations.

Thanks you !

I have a (maybe silly) question. Is it expected that when mapping tasks, the flow exists when the first element of the map succeeds (i.e. the first branch of the reference task)? That seems to be what’s happening to me, but perhaps I’ve missed an implementation detail. As a followup, if I want to wait for all mapped results, do I have to make the final reference task a reduction over upstream mapped results?

Hi there I'm facing again this error with the current version 0.15.5:

requests.exceptions.ConnectionError: HTTPConnectionPool(host='host.docker.internal', port=4200): Max retries exceeded with url: / (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7f43cca967c0>: Failed to establish a new connection: [Errno 111] Connection refused'))

First time i got this was in 0.14.19, but not in 0.15.4.

Hello everybody! My colleague (@Greg Roche) and i tried to update our self hosted prefect server (on EC2) today (from 0.14.15 to the latest version - 0.15.5). After updating prefect we tried running the server but we saw already some errors in the console. We were still able to connect to the UI and we saw that all the connections to

:4200/graphql

were refused. After some trying we gave up and downgraded back to 0.14.15 again (it worked again but we lost the data - flows, flow runs etc.) - no big deal. I was curious and thought if connections were just refused it shouldn't be too difficult to get it running - so i created my own

docker-compose.yaml

file from

docker prefect config

and changed some values. I ended up changing the interfaces that the services are listening on (graphql,apollo,ui) to 0.0.0.0 - after that it worked again. This makes it impossible for us to update our prefect server 😞 I couldn't find anything in the

prefecthq/server

repo that could cause those connection problems (at least from the most recent changes) - could someone please look into this? If you need any more information please let me know 👍 Thanks in advance!

Hey prefect community, I’m building a prototype of an ETL pipeline that is supposed to be run on dask/kubernetes. Let’s assume the dummy ETL pipeline is supposed to download 50k images from a bucket, rotate them, and write them to another bucket. To do so, I created three separate tasks:

@task
def load_image_from_bucket(fn):
    # Load image img from filepath fn in bucket
    return img, fn

@task
def rotate_image(x):
    img, fn = x
    return img.rotate(90), fn

@task
def save_image_to_bucket(x):
    # Save rotated image to new bucket

The flow looks as follows:

with Flow(
            'Load images from bucket',
            run_config=KubernetesRun(),
            executor=DaskExecutor(...)
          ):
    # Get list of image uris `imgs`
    loaded_images = load_image_from_bucket.map(imgs)
    rotated_images = rotate_image.map(loaded_images)
    save_image_to_bucket.map(rotated_images)

According to the docs, this should create parallel pipelines (without reducing after getting e.g.

loaded_images

):

However, if a mapped task relies on another mapped task, Prefect does not reduce the upstream result. Instead, it connects the nth upstream child to the nth downstream child, creating independent parallel pipelines.

From what I observe when using the `DaskExecutor(…)`and applying the above flow to 500 images, all workers only download images at first and only start rotating once all have been loaded. It either takes multiple minutes until they start rotating and uploading but often the flow just crashes before the flow reaches that point. I would want depth-first execution: a worker downloads an image, rotates it, pushes it to the bucket, forgets about this image and moves on to the next one. This appears to be (roughly) the case when running with the following config using

executor=LocalDaskExecutor()

instead of an ephemeral dask cluser in k8s:

with Flow("name") as flow:
   ...

state = flow.run(executor=LocalDaskExecutor())

Transforming 500 images in a bucket then takes only ~50s. Is there a way to influence the order in which the tasks are executed when using the DaskExecutor with an ephemeral dask cluster in k8s? Thanks a lot!

I’m experimenting with a preprocessing pipeline using a DaskExecutor using an ephemeral dask cluster within a k8s cluster. I’m seeing this warning:

//anaconda3/envs/sandbox/lib/python3.8/site-packages/distributed/worker.py:3862: UserWarning: Large object of size 1.65 MiB detected in task graph:
  {'task': <Task: load_image_from_bucket>, 'state':  ... _parent': True}
Consider scattering large objects ahead of time
with client.scatter to reduce scheduler burden and
keep data on workers

Is it a bad idea to pass e.g. images around between prefect tasks?

When I run a flow without using all the parameters I get an error Flow:

with Flow("ROI Report Generator") as flow:

    start_date = Parameter(name="start_date")
    end_date = Parameter(name="end_date")
    to_list = Parameter(name="to_list")
    email_domain = Parameter(name="email_domain", default="")
    account_uuid = Parameter(name="account_uuid")

Run:

state = flow.run(
            account_uuid=527132950,
            start_date="2020-11-01",
            email_domain="",
            end_date=None,
            to_list=["<mailto:jeff.newburn@logikcull.com|jeff.newburn@logikcull.com>"],
        )

Error:

TypeError: run() got an unexpected keyword argument 'account_uuid'

Can someone help me understand what is going on here? Why do Parameters have to be used in a flow for the flow not to explode?

Hi there, what is the best way to trigger a flow from the state of another flow/s within a flow of flows? I.e, given flows

a

,

b

, and

c

started within parent flow

p

using

startflowrun

, how can I trigger (or not trigger) flow

c

, given the states of flows (startflowrun tasks)

a

and

b

? Within flows

a

and

b

, there is a task to skip downstream tasks (raising a SKIP signal) based on a condition. Thanks in advance.

Hi guys, I am trying to understand a specific behavior when flows are being built and sent to the server. In one of my tasks I am defining the URL for Livy as follows

livy_url = f"https://{host}:{port}/gateway/default/livy/v1"

host and port are referring to variables that have the following code:

os.environ.get('LIVY_HOST')

and

os.environ.get('LIVY_PORT')

the livy_url is getting the variables from my local machine when the flow is being built. and thus the values for host and port are incorrect.

Hey folks! I'm new to Prefect and I'm sure I'm doing something dumb. I have a very simple task (like a few simple string manipulations) that I'm mapping over about 4000 list items, and I'm noticing that Prefect takes around 5 minutes to complete the tasks, is causing about 500MB of network traffic, and postgres is writing 600MB to disk. I'll provide more info in the thread below. Any advice would be much appreciated.

Hello friends! I'm trying to implement the scenario of the prefect's work with a dask cluster of 20 ~ 30 workers and it won't work. I'm asking for the best practice. I have the following tools: Debian server with a running prefect version 15.5 and an installed dask-2021.9.0 distributed-2021.9.0.  I run dask-scheduler on the Debian server and one of the dask-workers on another computer. Details are in my messages

I'm trying to run prefect in the docker container, using a local agent for it but the flow run does not complete the execution. In terminal get these logs, it goes as

deploying-->completed deployment-->returned with exit code -11!

, Attaching the screenshot. What is the

exit code -11!

here and how can I solve this issue.

Hello, guys. I have a KubernetesRun with image

image=f"project_name/prefect:etl_tools-{sha}"

where sha is a git commit SHA. And I have a registration bash file that call

prefect register --project etl -p "/flows/etl/**/**.py"

. Obvious that every commit to repo change SHA, change the image, and registration process bump version for all flows. There is a way to NOT bump all version and exclude image name from the idempotency key?