Hello! does Prefect support asynchronous tasks?

Hi, how would I change the

PREFECT__CLOUD__AGENT__AGENT_ADDRESS

field fo Kubernetes deployment without modifying the yaml file directly? Is there a command line argument that I can use? I was suspecting

--agent-address

, but that’s not available.

- name: PREFECT__CLOUD__AGENT__AGENT_ADDRESS
          value: http://:8080

Below is the command I’m testing

prefect agent kubernetes install --backend cloud --api "<https://api.prefect.io>" --namespace xxxxx --tenant-id xxxxx --service-account-name xxxx --key xxxx --rbac | kubectl apply --namespace=xxxxx -f -

Is it possible to increase the TTL from the idempotency key?

Hi there - I have a flow where I would like to (at flow run time) get a date (as a string) from a function defined in my codebase and provide it to ALL of the tasks in the flow. This sounds like a Parameter, but there are a couple problems with this: 1. a function call is not JSON serializable 2. I don’t really want every single task to depend on one node that is just getting a string - the graph view would be extremely cluttered Any ideas on how to accomplish this?

are there any limits of what should be stored in a prefect context? I.e. is it a bad idea to store a dataframe in prefect.context so that I can pull this dataframe from a state_handler? My alternative (I think) would be writing the dataframe to disk, then setting the location of that dataframe in the context and then pulling the dataframe location from context in the state hanlder.

Hey Community, As we have been scaling out our prefect usage, we are starting to be quite hampered by a few limitations with how prefect might be handling ECS and its fargate task registration. I keep running into this error:

an error occurred (ClientException) when calling the RegisterTaskDefinition operation: Too many concurrent attempts to create a new revision of the specified family

I have followed the instructions here, but have had no change in the behaviour. We run about 2500 flows every 24 hours and they are generally short running ~60 seconds and run every 15 or so minutes. I use the aws cdk and this is how I have injected the variables into my fargate ecs agent.

fargate_task_definition.add_container(
            id="prefect-fargate-agent-container",
            container_name="prefect-fargate-agent-container",
            image=ecs.ContainerImage.from_ecr_repository(
                prefect_agent_image_repo, "latest"
            ),
            cpu=256,
            memory_limit_mib=256,
            logging=ecs.AwsLogDriver(stream_prefix="prefect-fargate-agent-container"),
            environment={
                "AWS_DEFAULT_REGION": "ap-southeast-2",
                "LAUNCH_TYPE": "FARGATE",
                "AWS_RETRY_MODE": "adaptive",
                "AWS_MAX_ATTEMPTS": "100",
                "TASK_ROLE_ARN": task_role.role_arn,
                "LABEL": "prefect-agent-fargate",
                "NAME": "prefect-fargate-agent",
            },
            secrets={
                "PREFECT__CLOUD__AGENT__AUTH_TOKEN": ecs.Secret.from_ssm_parameter(
                    prefect_cloud_agent_auth_token
                )
            },
        )

Is there anyone who might be able to help me fix this issue, its really hurting us at the moment. Cheers

Hi all! new member here, so hoping not to be asking an already answered question! Q: How do you set up a project folder structure that allows locally developing and registering multiple flows? If using a single script (not a package) having

if __name__ == '__main__':
    # run/register flow

works fine. But if you have modules python does not like it. I was hoping for a structure like the screenshot attached below. For now, I import the flow I am working on the

main,py

file and that's the one I use to register my flow and test locally (with docker-compose, local agent and the UI)

Hi, Anyone knows how to name task runs based on inputs, but without using the

@task

decorator? I am sub classing

Task

and would like to name it based on the inputs given to

run

Is there a way to retrieve a full list of users on a Cloud tenant? I couldn’t find anything on the GraphQL docs

Hi, how can I pass environment variables to Docker agent without restarting? is there any options to set it in flow, for example pass the path for *.env file or make visible variables from current terminal session to Local agent?

Can I make tasks read from a local folder as a cache using

snapshotting

?

Hey guys, I`m new working prefect (love it so far bowtie). Our plan is to use it as a job scheduler for our dwh. In general we are planning to do ELT with Fivetran and dbt, therefore we are also looking into dbt cloud. My first idea was using pyhton request library to simply trigger via api the dbt job, but I`m wondering if you guys have any smarter idea or what's the prefect-isch way to solve the task. I also looked into the dbt shell execution, but does it also work together with dbt cloud? Thanks in advance, Ben

Hi there. I recently started getting some errors on VS Code using pylance. For example, the following line

from prefect import Flow, Parameter, task, unmapped

produces

"Flow" is not exported from module "prefect"
Import from "prefect.core" instead Pylance(reportPrivateImportUsage)

"Parameter" is not exported from module "prefect"
Import from "prefect.core" instead Pylance(reportPrivateImportUsage)

"task" is not exported from module "prefect"
Import from "prefect.utilities.tasks" instead Pylance(reportPrivateImportUsage)

is not exported from module "prefect"
Import from "prefect.utilities.edges" instead Pylance(reportPrivateImportUsage)

Also, I am getting pylance errors when passing Parameters to tasks that expect

str

arguments. I don't believe this was happening before. Do you know if something changed recently in pylance or in how we are supposed to import modules? I am using prefect 0.15.4 Thanks!

I understand that for local execution, we can set local secrets; is the same possible for the key-value store? My unit tests are failing because I don't auth to Prefect, so I'm wondering whether I should mock

get_key_value

or if there's an ENV I can alternatively set.

I am hoping to have a task be blocked by an upstream task whose execution depends on a parameter, and that does not necessarily provide any input to the downstream task. With my current flow structure the downstream task is only executed when my parameter is set to

false

. How can i get it to execute in both cases where the parameter is

true

or

false

? Sample code in thread

Can I look at a prefect flow run and tell what image it used in the

KubernetesRun

? The name of the image or ideally its hash?

I’m trying to understand how

KubernetesAgent

(code here) uses the environment variables used in the k8s deployment, particularly

IMAGE_PULL_POLICY

, to build the podspec for a flow run. It looks like if a flow specifies

run_config=KubernetesRun(…)

at all, then

IMAGE_PULL_POLICY

from the deployment env is ignored? Is that right?

related to this thread, I’m trying to set a key in the context in one task, and then read it in a subsequent task. Is that something that should work? Here is the task that sets the key:

class SetTrDate(Task):
    def run(self, tr_date: str):
        prefect.context.tr_date = tr_date or get_last_tr_date()

and then I read it in subsequent tasks with :

prefect.context.get('tr_date')

The subsequent tasks seem to think that it’s set to

None

Should this work?

new to Prefect here, but how do we avoid crash-looping when using an

IntervalSchedule

if the workflow is raising an exception (so we'd like to simply cancel the workflow as soon an error occurs)?

Hello folks! Is there anyone to cancel a scheduled flow from the prefect cli command?

hey =] couldn’t find a clear explanation in the docs - what could be a reason for me to get a

task is already running

(for a KubernetesRun)? and then

Finished task run for task with final state: 'Running'

…. (other than that, it seems like the task doesn’t really run.

I seem to be having an issue when registering a flow with the cloud. It fails here:

File "/usr/local/lib/python3.9/site-packages/marshmallow/fields.py", line 1410, in _serialize
    return int(value.total_seconds() / base_unit.total_seconds())
AttributeError: 'int' object has no attribute 'total_seconds'

In the comments I have the full stack trace

If you return a python set from a FunctionTask and try to use it in another task B, Prefect fails B silently without reason.

Problem in running ecs agent: i am running an ecs agent as followed:

prefect agent ecs  start --key $KEY --task-role-arn $TASK_ARN --log-level INFO --label s3_sync --name farget-dev --execution-role-arn $EXEC_ROLE_ARN

and its giving me an error:

ValueError: Failed to infer default networkConfiguration, please explicitly configure using `--run-task-kwargs`

I am created a ecs cluster (using default). I used following command to use create cluster:

aws ecs create-cluster

which creates a default cluster. Can anyone point out if i am missing anything?

i see this is the default task_defination template : https://github.com/PrefectHQ/prefect/blob/master/src/prefect/agent/ecs/task_definition.yaml But i am not sure if i need to pass anything or create a task definition template.

Hello everyone. I trying to figure out what is considered best practice for differential dataloads and backfills? I am trying to obtain the timestamp of the last successful run of a flow in order to obtain data added or changed from last run. Using the prefect

Client.get_flow_run_info

-function seems to be one way of achieving this, but it may not scale as number of runs for a flow grows large.

Maybe a stupid question, but can anyone tell me what the Blue/Green colouring means on the DAG overview diagram?

Hi ) How do I make sure a Prefect flow which is scheduled to run each 10 mins is NOT run if the previous run is still in progress?

Hi Guys! Im getting this error when using the UI. Im running a server and an agent in a docker container. {"type":"http-log","timestamp":"2021-09-20T143829.787+0000","level":"error","detail":{"operation":{"user_vars":{"x-hasura-role":"admin"},"error":{"path":"$.selectionSet.flow_run.args.where.flow.project_id._eq","error":"unexpected null value for type \"uuid\"","code":"validation-failed"},"request_id":"0b7c4c08-10ce-4782-9b6b-4a2e3ee547bf","response_size":142,"request_mode":"error"},"request_id":"0b7c4c08-10ce-4782-9b6b-4a2e3ee547bf","http_info":{"status":400,"http_version":"HTTP/1.1","url":"/v1alpha1/graphql","ip":"172.21.0.6","method":"POST","content_encoding":null}}} Any suggestion?