Hey dear community We recently have an issue when we were triggering 3 instances of the same parent flow that triggers a child flows (it's not the common use case but we wanted to test this case also) And we get errors from the prefect-server that goes like this.. - I will write in the thread What can possibly be the cause? Thanks

Hello everyone, I'm trying to launch a task through the prefect with a call to clickhouse, I did not find the library of the prefect's tasks. Can anyone who has done a similar task, please share the solution, thanks

Hey guys a quick question I intend to create a task that would use

awswrangler

to save my dataframe to s3. my environment is already

awscli

compliant, do I need to go through prefect Storage? Secondly if I'm doing all this with aws ec2 remote instance how do I activate my flow. Can I do on the command line:

python myfirstprefectflow.py

? Assuming I have already indicated a schedule in my flow code ? Or do I use crontab to schedule again 🤔. I would be glad if I could be pointed in the right direction on using aws ec2 to run prefect.

Hello, beginner's question here! I see a lot of examples that are like this:

with Flow("my-flow) as flow:
    # do some things here
# EITHER flow.register("project_name")
# OR flow.run(executor=DaskExecutor())

I want to run my flow through prefect cloud, so I understand that I need to register my flow , but when I want to run it it looks like I need to edit the code, which seems a bit weird when it's dockerised and running on a remote server. Is it normal to register the flow like above before building the docker image, then switch it to the run command before building? Or is there something I'm missing?

HI! I am getting an error*:* 

ModuleNotFoundError("No module named 'parameters'")

root
  - src
    - flows
      - flow_1
        - flow_1.py
        - parameters.py

I am registering the flow with this

bin/prefect register --project «myProject» --path ./src/flows/flow_1/flow_1.py

When I run the flow locally with 

flow.run()

 it works fine 🤔

Hello all I am attempting to get information from the current flow that I'm in using GraphQL, however, I'm only getting an empty response from the API. I'm almost dead certain that the query is correct as I've used it in the sandbox first. We're using Server for our flows. Any help would be appreciated. Thanks

FYI Slack has some issue - if you can't post your message here, you can use Discourse or submit a GitHub issue

Is there anybody out there??😞

Hi guys, I want my flow to not move onto the next task if one of my tasks is retrying.

@task(retries=2, retry_delay_seconds=10)
def task_1(time_to_sleep: int):
    raise Exception()
    time.sleep(time_to_sleep)


@task(retries=3, retry_delay_seconds=10)
def task_2(time_to_sleep: int):
    time.sleep(time_to_sleep)


@flow(name="steps_flow_dask", task_runner=DaskTaskRunner(address={dask_address}))
def flow_to_run_tasks():
    task_1(3)
    task_2(2)

Currently when

task_1

throws the exception

task_2

immediately executes. The behaviour I want is that

task_2

doesn't execute unless

task_1

is successful. Also, if

task_1

hits its' maximum number of retries then

task_2

should not execute. Any ideas? This is on Orion.

Hey all, we’re seeing some unexpected / buggy behaviour with Prefect Cloud, hoping you could advise? We have a flow with a daily cron schedule which gets redeployed once a day (as code changes are made). Previously when the new version of the flow is registered (part of our deployment), the old version of the flow becomes archived and the schedules for that old version were cancelled, but over the last week we’ve been seeing double-scheduled flows remaining - one scheduled flow_run for the current flow, and one for the previous (now archived) flow. Wondering if you’ve seen this issue elsewhere? In addition we’ve seen some (likely) related oddities with our Automation which triggers a PagerDuty incident if the flow doesn’t start within the first 5mins. It looks as though these phantom archived–but-still-scheduled flows are trying to run, hitting our concurrency limit of 1 and stalling, triggering PagerDuty, but then when we click through on the flow_run link, the flow run literally doesn’t exist (either in the UI or in GraphQL). It’s super weird. Hope someone can advise! We’re the

Deliveroo

prefect tenant, happy to send over some specifics / URLs if helpful

Hi all, Is there any special flags or procedures for when deploying agents in servers that connect to the internet through a proxy? I have this wired behavior that I been fighting with since yesterday. I start the agent normally from CLI (Docker Agent) and It’ registers successfully (it even shows that in the Cloud UI). When I try to quick-run a flow from the Cloud, the flow is not being picked up by the agent even though the labels are exactly the same. in fact in the Agents section message about “Late submittable runs” starts to appear. Maybe some kind of caching in the proxy? Thanks in advance.

Hi all, I’m trying to use a secret in a state handler but it doesn’t seem to be working

Hey All, I’m having an issue with a flow that is registered with the Hybrid model approach where I have a docker image and the flow with a DockerConfig to run it from the cloud. This was working in a the past but as of a few days ago, flow run gets triggered but no run is being performed. This leaves the run in Submitted state with a few attempts to re-run and Lazarus service finally kills it. I have other flows registered/installed and working within the same environment so I a little puzzled as to why this flow does not run. Any debugging suggestion would be appreciated.

If a task run can be in a “Running” state multiple times in its lifetime, should there be an identifier that uniquely identifies a particular “run” of a task run? We could take task run ID + task_run_count to come up with an identifier for this, but I’m wondering if Prefect has thought about this or plans to support this in the future?

Hi all. I have a couple of ETL flows that are going to run on a large quantity of data the first time they run, but much smaller datasets every subsequent run. What I want to do is run with a DaskExecutor on the first run, but a default executor on subsequent runs. I was wondering if it's possible to choose an executor at runtime based on a parameter, or if there is another way to achieve what I'm looking for without having multiple flows. Thanks!

Is there an alternative/direct way to get all flow runs via flow id besides:

runs = FlowRunView._query_for_flow_run(where={"flow_id": {"_eq": id}})

I'm running flows on ECS using fargate; is it normal to see

RUNNING

tasks like those shown below which have been running for days? Are those truly still running? Am I being billed for idle compute here?

Hi all, I am running into an issue when running a flow where the repository is no longer accessible when upgrading to

0.15.12

from

0.14.21

Failed to load and execute Flow's environment: NotGitRepository('No git repository was found ...

I am using the git storage interface and have no issues when using version

0.14.21

Stupid question, is there an equivalent of

prefect run

in Orion? Or do you have to run flows with

python flow.py

?

Hey all, I'm getting the following error when I try to run a flow via cloud using Bitbucket storage config:

Failed to load and execute Flow's environment: ValueError('No flows found in file.')

. The flow context manager is defined within the body of a function that returns a flow, I'm wondering if that is the reason why cloud can't find the flow?

Does the DbtShellTask work with the open-lineage dbt? (

dbt-ol run

)

Hey Prefect community! I’m using ECS task runs with Github storage for my flows but running into the following error when running the task:

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named 'constants'")

I’m importing constants which will be shared across many flows: the project name, ECS cluster name, Github Token secret name from a

constants.py

file in the same directory. Running the flow locally with

flow.run()

works just fine as expected, but when running it in ECS

constants

is not found. How do I register other module dependencies with Github storage, if possible? Or do I have to add these constant values to each of my flow files? Thank you! Example below:

. (repo root)
|____flows
| |____hello.py
| |____constants.py
| |______init__.py

Is there any documentation on having a main flow and sub flows that can branch off from the main one?

Hi everyone, is there a way I can use FileHandler or other means to write log to a log file when I use Prefect Cloud + Local Agent? The FileHandler works when I run locally with flow.run() but it writes nothing if I register the flow and run it on Prefect Cloud.

Hey all! Has anyone here needed to "package" regular files with their flows? We can, of course, upload "file packages" somewhere else and download them each time a flow runs, but specifically for the

DbtShellTask

, I'm thinking it would be nice if a flow we register and store on s3 also came with its dbt project directory. Currently, I don't think it's possible and will store the "packaged files" on s3, but I also don't want to start that work if there's already a solution that I haven't seen. 😄

I'm trying to implement flat-mapping in a flow, but the final task doesn't seem to run the correct number of times. Code in thread.

Hi all! Does anyone have experience with controlling BigQuery reservations (Flex slots) from Prefect? Any best practices to share? I'm especially wondering how to ensure slots are canceled after a run. My basic use case would be to lower our BQ costs on our dbt runs: 1. Purchase flex slots 2. Run dbt models 3. Cancel flex slots Thanks for your input!

Quick Question, I have recently seen an issue where the Agent does not pick up the triggered flows. The Prefect Server and Prefect Agent are both running & the labels on the Flow versus the Agent match, so that is not the issue. Is there anything obvious that I need to check? Appreciate the help. I am running on version 0.14.8

Hi ! I have the following flow :

with Flow("sec daily", state_handlers=[flow_handler], result=GCSResult(BUCKET_RESULTS)) as sec_daily_flow:
    root_dir = init_dirs("sec", archive=False)
    download_dir = download(root_dir, False)
    metadatas = metadata(download_dir)
    upload(metadatas)
    result_locations = split_store_batches(metadatas)
    mapped_flow_run_ids = create_flow_run.map(
        flow_name=unmapped("index_flow"),
        project_name=unmapped("Document Pipeline"),
        parameters=result_locations
    )

With a "subflow":

with Flow("index_flow", result=GCSResult(BUCKET_RESULTS), state_handlers=[flow_handler]) as index_flow:
    # Imports
    metadatas_batch_location = Parameter("metadatas_batch_location", required=True)
    metadatas_batch = get_result(metadatas_batch_location)
    imports = build_import(metadatas_batch)
    index_weaviate(imports)

But I have the following error :

prefect.exceptions.ClientError: [{'path': ['create_flow_run'], 'message': 'dictionary update sequence element #0 has length 1; 2 is required', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

That I'm not sure to understand 😕 I need some help please 🙏 Same error for both 'batches': • Task 'create_flow_run[0]': Exception encountered during task execution! • Task 'create_flow_run[1]': Exception encountered during task execution!

Hello everyone, does someone knows how to disable the SSL Verification when using prefect? Is there's something like the "*--set ssl_verify True*"? I'm facing some problems with auth and running the agent due to a corporate network. The following error occurs when I use the "*prefect auth login --key*":

requests.exceptions.SSLError: HTTPSConnectionPool(host='<http://api.prefect.io|api.prefect.io>', port=443): Max retries exceeded with url: / (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1129)')))