I have a flowrun with some curious logs. The logs claim that the heartbeat was not detected and the flow was cancelled, but according to the server it was still running and I cancelled it manually just now. The id is 99b7bd01-7b90-4634-9718-8e2eab1c00b0. Thank you.

Still learning Prefect concepts here and would love some advice. I'm trying to create a flow where I map a bunch of tasks. then when some of them fail, run a recovery task for each failing task, passing it the input of the original task to do an alternative action. At first it seemed like employing state handlers would help, then Kevin pointed me towards using the "any_failed" trigger instead, but it seems like that is built for a scenario where you have X tasks and you want to run 1 recovery task- but I have X tasks and want to run Y recovery tasks (somewhat less than X). It seems from the docs that prefect caches the inputs to a task, but I don't see how to access those inputs in a subsequent task. Is this possible? Thanks!

It seems Orion has an issue with the newest version of 'click' (8.1.0): "ImportError: cannot import name 'get_terminal_size' from 'click.termui' (/home/ubuntu/orionenv/lib/python3.8/site-packages/click/termui.py)". Downgrading 'click' to 8.0.3 (the version that I had elsewhere) made this error go away.

Hey all! Don't forget to join @alex on the Great Expectations great expectations Community Call starting at 12p Eastern: https://www.addevent.com/event/ag12883115 Posted in #events

Hello, can anyone help me? thanks 😄, I'm trying to create unit/integration tests to check my flows and tasks. I'm using Prefect Orion

prefect = "2.0a13"

, I did read this link, but I'm using async flows. More infos in the thread.

Hey gang. Today I released this new GitHub template repository. It includes all the bits necessary to deploy a simple application to Google Cloud and Prefect. Once the configuration is complete, you can ship changes with a simple GitHub release. https://github.com/biglocalnews/prefect-flow-template. There's a full tutorial in the wiki user manual.

All feedback welcome.

Hi folks. I'm trying to get my tasks to log to Cloudwatch through the ECSRun configuration. I'm testing out specifying a yaml config in

task_definition_path

, and running into an issue. My flows run on different ECR images, so I don't want to specify the image in the Task Definition file. However, when I leave it out, and specify it in the constructor:

run_config = ECS_RUN(image=ECR_IMAGE, task_definition_path='s3://<path>'

I get an error that container.image should not be empty. My task definition file that I'd like to use is:

networkMode: awsvpc
cpu: 1024
memory: 2048
containerDefinitions:
  - name: flow
    logConfiguration:
      logDriver: awslogs
      options:
        awslogs-group: test_prefect_log_group
        awslogs-region: us-east-1
        awslogs-stream-prefix: ecs-prefect
        awslogs-create-group: "True"

Is there any way to use a Task Definition file without specifying an image? Or is there another/better way to make sure my ECS tasks get logged? Thanks for any help.

Hi, trying to print out the

date

and

today

context in a task, but getting None.

date = prefect.context.get("date")
today = prefect.context.get("today")
def get_data():

    print(f"{date}")
    print(today)

hi all, i'm trying to write a flow where all the tasks should execute in a sequential order but some of them should optionally be skipped. basic example in normal python:

print("hello world 1")
if not skip_task2:
    print("hello world 2")
if not skip_task3:
    print("hello world 3")
print("hello world 4")

here's an attempt in Prefect:

@task()
def print_task(value: str) -> str:
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(value)
    return value


with Flow("case_skip_flow") as flow:
    skip_task2 = Parameter("skip_task2", True)
    skip_task3 = Parameter("skip_task3", True)

    task1 = print_task("hello world 1")
    with case(skip_task2, False):
        task2 = print_task("hello world 2", upstream_tasks=[task1])
    with case(skip_task3, False):
        task3 = print_task(
            "hello world 3", upstream_tasks=[task1, task2], task_args={"skip_on_upstream_skip": False}
        )
    task4 = print_task("hello world 4", upstream_tasks=[task3], task_args={"skip_on_upstream_skip": False})

i'm getting into trouble with the upstream skip status for

task3

. • if I set

skip_on_upstream_skip

to

True

, then when

task2

is skipped, so is

task3

regardless of the value for

skip_task3

• if I set

skip_on_upstream_skip

to

False

, then

task3

runs regardless of the value for

skip_task3

how can i implement a pattern like this?

Hi, I have a flow that contains a few docker related tasks (amongst many others) - PullImage, StartContainer, etc Works fine locally but proved to be tricky to run as a DockerStorage on a test vm. After some trial and error the solution I found how I can start a container from a container (in case someone else is stuck too):

flow.run_config = DockerRun(image="my_image", host_config={"binds":["/var/run/docker.sock:/var/run/docker.sock"]})

But I also need to pull an image when its not available and this step fails since I am not authenticated from inside my flow container. I can probably create a shell task that does docker login for me before image pull, but I am wondering if there are better options?

Hello, we have a prefect 2 server running a on GCP Cloud Run service - could you point me at some docs for how auth is setup, are there any options other than identity aware proxy for example (ideally for both front end and command line access)

Hi all, quick question: I can't find a mongodb crud task. Have any of you guys have something that works?

Hello all. Have a small question about cloud hooks. According to the documentation they work only at the flow level and not at the task level. One of the states I can select in the cloud hook is "TimedOut". However I can't find how to get my flow to enter that state. When I create my flow, I don't see any timeout condition for the entire flow run itself. Have I understood it correctly that cloud hooks only work at the flow run level? And if so how can I get my cloud hook to trigger when a flowrun enters the state "TimedOut"? Thanks in advance

Hi everyone. Am using a github action to register a flow. The flow uses S3 as storage and run_config is fargate config. The problem is, when registering it fails with an error related to boto3, which am not even using boto3, but before it was failing due to lack of boto3 library, so I added a command to install boto3 with pip. But now the error is "botocore.excwptions.ClientError: An error occured (AccessDenied) when calling the putObject operation: Acess Denied" Is this setup feasible? Why is it asking for boto3? Does prefect use boto3 library to use S3 as storage in the flow? Should I add auth to aws using github action or aws auth using boto3 in the flow python code itself?

Hi Everyone! I have a small question regarding dedicated execution of prefect tasks using a DaskExecutor. For example let's say I have some special tasks in a flow that need to be executed specifically on a dedicated machine. Is it possible to provide this information in Prefect?

Hey folks! We have a single agent running on EC2 and it has a few tags associated with it, some of which are associated with a flow concurrency limit. (e.g. etl_jobs => limit:5, ml_jobs => limit:2, and so on). Am I reading the Prefect 2.0 docs correctly? We would need to run multiple agents on our existing EC2 instance if we want to have different Flow concurrency limits for different groups of flows? We would have an “ETL Job” work queue with it’s won limit and a dedicate agent, another work queue for “ML Job” and a related agent.

In Orion, task run result storage is not implemented yet. Is there already a blueprint on what it will look like?

Does anyone know if it’s possible to find out the ECS task ID of a Prefect task from Prefect or the task itself? The context is I am using the ECSAgent to run a task which dumps its logs into a cloudwatch group which is called

prefect/task/{ECS-task-id}

. My aim is to display those logs in our webapp to show admin users the status of the flows they’ve started Ideally I’d be able to use the ID of a flow run and use that to query the ECS task it ran on and then use that to query the logs of the flow

I installed a dask cluster (https://docs.prefect.io/orchestration/recipes/k8s_dask.html#kubernetes-manifests) on our kubernetes cluster and I can run flows on it no problem using flow.run(). However, when I register the flow it tries to run the job as a kubernetes job. I have prefect server deployed and can see the runs, but they all fail with 'module C not found'. I am missing something, why do the flows not execute on the dask cluster?

Hi! I want to be able to set some Parameter when I register flows (or I guess replace the default value), is it possible?

Hi, Orion question, is there any upcoming plan in the roadmap that allows modifying the parameters of a deployment from the UI before launching a Flow Run of it. Thanks!

Hello, I am running into what seems like a simple problem, but not seeing how to do it. I have an implicit dependency from taskC to taskD, but can't get prefect to understand it

Hey guys is possible to hide the "RUN CONFIG" tab from Prefect Server? Because today everyone can see the secrets.

I’m running into an issue trying to use GitLab storage to reference flows stored in my gitlab repo. This is how I have it in my script:

from prefect.storage import GitLab
storage = GitLab(repo="org/repo",
                 path="/hello/hello_k8s.py",
                 ref="prefect")

And when I try to run the flow, I get 404 error file not found but the file is there. Is there a way to print out what path prefect runner is using?

I am trying to run a simple terraform code using prefect. the terraform code is a .tf file which creates a storage account. When I am running it as a normal python code (without including prefect, flows etc), it is working fine. But with prefect I am getting error. I am not sure if this is related to prefect, or something else.

👋 hey all! You probably saw the module that was published to the Terraform registry recently, that builds the Prefect Docker agent and related infrastructure. Come join @Jamie Zieziula, @George Coyne and myself on PrefectLive in 15 minutes as we walk through that module, how it was built and how to use it! twitch.tv/prefectlive

Hi everyone! Does someone have examples of how they’ve used the SnowflakeQuery task?? Specially interested in examples using the imperative api! I have the following so far:

flow.set_dependencies(
            task=SnowflakeQuery,
            keyword_tasks=dict(query='''SELECT * FROM dummy_table;'''
            )
        )
flow.set_dependencies(
            task=save_query_result_as_df,
            keyword_tasks=dict(result_set=SnowflakeQuery)
        )

The task

save_query_result_as_df

currently only does a

print()

of the

result_set

. I’m confused about a few things: • I didn’t provide the

SnowflakeQuery

with any authentication. Why doesn’t it throw an error about that? • The print statement prints out

<class 'prefect.tasks.snowflake.snowflake.SnowflakeQuery'>

, which makes me think the snowflakequery task wasn’t run? If it wasn’t run it would at least explain why it didn’t throw an error 😅 but how can I make it run?

I'm running into some memory limits with a flow using an ECSRun config on a LocalCluster DaskExecutor. The first task in the flow is to generate a list of file keys, and then I map over that list to process the files. I would have expected that when an individual Mapped Child task completes, any memory used in that task would have been released (aside from the return value). But it seems like that is not the case, as the process eventually runs out of memory after several Mapped Children finish. Is there something I can do in the configuration of the flow or task to mitigate this? As an aside, when I run this same flow using a FargateCluster, it will complete without issue. I'm not sure if that's because the FargateCluster I'm using just has more memory, or if there's something inherently different between the two different DaskExecutor classes. Thanks as always for any help you can provide.

If you haven't already, join us on PrefectLive right now! 🔥 covering infrastructure topics such as deploying Prefect agents with Terraform with our super-knowledgeable Senior Infrastructure engineers! The demo starts now! 🙂