Hi, I am trying to create a flow based on some dynamic values. These dynamic values tell the tasks which are required to create that flow. My tasks are saved in separate files. For eg: task1.py

def task1():
    @task
    def run_terraform_lt():
        tf = Terraform(working_dir="law_tf")
        tf.init()
		tf.apply()

task2.py

def task2():
    @task
    def run_terraform_rt():
        tf = Terraform(working_dir="rt_tf")
        tf.init()
		tf.apply()

The input is similar to this:

{
    "task1": {
      "id": "foo",
      "task_name": "task1"
    },
    "task2": {
      "id": "bar",
      "task_name": "task2",
    }
}

I am getting the task name from this dictionary. Based on the value of "task_name", I have to create a flow combining all tasks. I am creating a flow where I am trying to add them, but the tasks are not getting added in the flow. Anyone has any idea on how this can be achieved? Here is the snippet of my flow creation code. It is inside a for loop through which I am extracting the task name and other variables.

Hi everyone! I’ve created a flow-of-flows which runs another flow in a loop, something like this:

with Flow('Parent Flow') as flow:
    
    dates = ['2021-01-04', '2021-01-05', '2021-01-06'] // a list of dates

    for i in range(len(dates)):
       //run child flow with i as parameter

When I check the Prefect UI it seems like the child flows are running in parallel (I’m using threads) which is in general great but in this case I’d like the child flows to be run sequentially. Is there a way to force the parent flow to run them that way?? 🤔

Hello community, I tried to connect Prefecta with my SQL server database, but I got this error. who has already had this error, or has an idea about this error?

Trying to ship logs from our containerized flows executed on EC2 to DataDog, I’m trying to figure out what is the best approach. As far as I can tell we can: a) Use the DataDog agent already on the EC2 instance to collect logs from the containers b) Add the DataDog agent to the flow images and write a custom logger to ship the logs to statsd c) Something else I haven’t thought about I think a. has the advantage that we don’t need to do any custom coding on each flow and we get all the logs for all flow runs for free. Anyone have experience with this approach? It seems all we need according to this: https://docs.datadoghq.com/agent/docker/log/ is to enable the Docker logs collections and maybe configure the logs in the Dockerfile.

Hello Everyone, We currently have a setup of Prefect Core Server running on our on-prem environment using Kubernetes. We are running our flows using a LocalDaskExecutor and we noticed that the logs for these flows do not show up on the portal. Only logs for flows that do not use a LocalDaskExecutor show logs. Is there a special configuration required for Dask to send the logs?

Hi folks, does sombody deal with so called business rules management systems (drools, camunda)? Can prefect be considered as an alternative to them? Maybe it good choice to build your own business rules management system on top of prefect? Drools and camunda are java-based and heavy. I look for some lighwight pythonic solution like stream/batch (btw does prefect is really quick in processing streams?) DAG management system with visualization of graphs and ideally with ability to separate conditional expressions (like if ... then ...) and other code.

Hi all, wonder of anyone can point me to an official task for pulling a git repo? I am wanting to use such a task in the context of running DBT - can I perhaps use something from the underlying Github tasks - don't see an explicit clone task...

I have prefect server 0.15.7 deployed. My flows were running fine a few days back. They seem to not execute properly now as for some of the mapped tasks the server is not able to set the state as Success/Fail. Here is the screen shot. I am running this on a yarn cluster. Any help will be appreciated.

Hi all - wonder if anyone has any experience running prefect flows on GKE spot instances? Is there any way of dealing gracefully with possible SIGTERM signals?

Hey all, I get this weird behavior once in a while - there are no logs in 1 of the flows I’m running. Most times it works fine and I get all the expected logs but sometimes only this - see the attached file. Am I missing something? Anyone have the same issue? This specific run had a bug (a python bug in the app itself that I couldn’t see..) Thanks

Hi everyone. We are using

wait_for_flow_run

to wait for a subflow run. The subflow is a long-running job that can take more than 12 hours and sometimes it times out. I noticed that

wait_for_flow_run

uses

watch_flow_run

which raises an exception if the flow has been running for more than 12 hours. The 12 hours timeout is hardcoded. See https://github.com/PrefectHQ/prefect/blob/afda99411f91582ad187bf33671268d8d3c3c2c0/src/prefect/backend/flow_run.py#L95 We plan to upgrade to 2.0 shortly after it's released. Is this timeout limitation for subflows also implemented in Orion?

I have a flow that I broke recently (bug in code). Now when I try to execute it in prefect cloud, There are 5 flows submitted, but nothing is happening to them. The agent we have is successfully picking up and running other flows. Do I need to clear the submitted flows? and how would I do so?

My prefect flow has few tasks which are running terraform. Is there any way I can see the detailed log of the task getting executed. for eg, when terraform runs, it displays the logs on console. is there any way i can see those logs? in the prefect console, i see a very basic log which doesn't tell me what is going on inside the task

Hi guys, is there any way to have multiple "instances" of a flow? We have a single flow, but use it for N users, we wanted it to be 1 flow per user for monitoring purposes, but we can't seem to find a way to create a new flow from an existing storage, only changing a parameter. Any ideas?

I have a strange error on a task, any one seen these before?

Task 'copy_from_s3_to_sftp': Exception encountered during task execution!
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/dist-packages/prefect/engine/task_runner.py", line 876, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/usr/local/lib/python3.9/dist-packages/prefect/utilities/executors.py", line 454, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "flows/k8s/my_flow_name.py", line 46, in copy_from_s3_to_sftp
SystemError: unknown opcode

Hey I am trying to upgrade my prefect agent that I have running on EKS, I upgraded my prefect package locally it is now version 1.2. When I try to apply the changes

prefect agent kubernetes install  -k {key} --mem-request 4G --mem-limit 6G --cpu-request 2 --rbac | kubectl apply -f -

Nothing seems to happen I just get the message:

deployment.apps/prefect-agent configured
<http://role.rbac.authorization.k8s.io/prefect-agent-rbac|role.rbac.authorization.k8s.io/prefect-agent-rbac> unchanged
<http://rolebinding.rbac.authorization.k8s.io/prefect-agent-rbac|rolebinding.rbac.authorization.k8s.io/prefect-agent-rbac> unchanged

Any idea’s on how to successfully upgrade my prefect agent?

Hot fresh dbt content from Prefectionist and all-around cool guy @James Sopkin for those of you looking to automate dbt exposures: https://prefect-community.slack.com/archives/CL09KTZPX/p1650385229289719

Strange behaviour with multiple

upstream_tasks

on a ShellTask This works:

brt_commands = create_brt_command.map(adoc_fp=updated_adocs)
        brt_commands_logged = log(item=brt_commands, desc="BRT commands")
        brts = shell_task.map(
            command=brt_commands, upstream_tasks=[tomogram_fps]
        )

This fails:

brt_commands = create_brt_command.map(adoc_fp=updated_adocs)
        brt_commands_logged = log(item=brt_commands, desc="BRT commands")
        brts = shell_task.map(
            command=brt_commands, upstream_tasks=[tomogram_fps, brt_commands_logged]
        )

Quietening down logs - if I'm mapping out to a lot of job, have a lot of params etc, it can be difficult to follow logs. Is there a way I can set logging such that I only see something if it's broken. If I

export PREFECT__LOGGING__LEVEL=INFO

I see:

python3 tmp/shell_task.py 
[2022-04-19 21:09:30+0100] INFO - prefect.FlowRunner | Beginning Flow run for 'My Flow'
[2022-04-19 21:09:30+0100] INFO - prefect.TaskRunner | Task 'MyShellTask': Starting task run...
[2022-04-19 21:09:30+0100] INFO - prefect.MyShellTask | lsto echo
[2022-04-19 21:09:30+0100] INFO - prefect.TaskRunner | Task 'MyShellTask': Finished task run for task with final state: 'Success'
[2022-04-19 21:09:30+0100] INFO - prefect.TaskRunner | Task 'problem': Starting task run...
[2022-04-19 21:09:30+0100] INFO - prefect.TaskRunner | FAIL signal raised: FAIL('Oh no!')
[2022-04-19 21:09:30+0100] INFO - prefect.TaskRunner | Task 'problem': Finished task run for task with final state: 'Failed'
[2022-04-19 21:09:30+0100] INFO - prefect.FlowRunner | Flow run FAILED: some reference tasks failed.

If I

export PREFECT__LOGGING__LEVEL=ERROR

I see nothing. Ideally I'd like to only see messages about broken stuff. Ideas?

What's a good starting area for the local dev story? We're deploying to a production ECS cluster with a decent Github CI/CD so far. Ideally a simple

Makefile

would spin up a few Docker containers that one could test flows on. How can I abstract between a local and ECS agent in my flows to allow something like a

env

var to swap between?

are there any best practices around testing flows/tasks in prefect 2.0? I see this page but something we're noticing that testing can be particularly slow on `flow`s, (sometimes taking ~1s to start up a each test) and it appears we always need to wrap `task`s inside of a flow to test them

Any idea what might be causing errors with artifact generation? I’m getting these errors in my task runs:

Error during execution of task: ClientError([{'path': ['create_task_run_artifact'], 'message': 'Task run <task_run_id> not found', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}])

Cloud anyone from the devs take a look at this? https://github.com/PrefectHQ/prefect/discussions/5440

Hey all! Come hang with @Kevin Kho and myself on PrefectLive 📺 today at 3p Eastern live on Twitch. He'll be showing off Modin, PySpark, Pandas and Fugue. We'll also probably have an invigorating ⌨️ mechanical keyboard debate and would love for you to join us! For event updates and reminders make sure to join the #events channel.

I'm getting some weird behavior regarding modules. I can correctly import my flow and the Config class from

projects

from both Python and Jupyter Notebook:

from projects.example.flows.hello_world
from projects import Config

But when I run the same import from Prefect, I get the following error:

pipenv run prefect run -m projects.example.flows.hello_world                    
No module named 'projects'

I have a flow that runs 121 mapped tasks - two of those tasks failed however the overall flow was nor marked as failed

https://prefect-community.slack.com/archives/CL09KU1K7/p1650463806923319 We're going live in 5 minutes on Twitch!

Hello all! Still a Prefect newbie, and have quick question that I wasn't able to find in the docs: Is there a maximum time limit for how long a flow can run? And does the length of a flow run affect cost, or is cost purely based on successful task runs?

Hi all, I’m evaluating Prefect as a data pipeline tech upgrade for our company. I’m putting together some pipeline examples to share with the team and I’m wondering what a good dev-test-debug-deploy loop looks like. The example flows I’ve found in the docs and the blog use

flow.run()

for local dev, then edit the code and switch to

flow.register()

. I don’t like editing the code to switch between local dev and deployment, and it doesn’t feel like your typical Python webapp dev loop. Is there a better way to do this?

I'm getting a weird behavior - I can correctly fetch the secrets when running a flow locally - but when it's built on Github Actions - after a step correctly logs into prefect with auth, I get

ValueError: Local Secret "AWS_ACCOUNT_ID" was not found.