I’m currently receiving user input through

model = Parameter('model_name')

. I want to announce the user input in Slack via my state handler, but when I do it displays as

<Parameter: model_name>

rather than the actual user input. Does anyone have any suggestions on what I can do to display the user input?

I have a task within a flow that returns a structure

val = my_task()

where val is some json thing. I also want to set that tasks downstream dependencies, though, which I had been doing with

val.set_downstream(othertask)

(although I was always a little confused as to why that would work). Well, now it's stopped working. What's the right way to do both of these things?

Got an Orion question: Will I be able to map subflows across a Dask cluster?

I'm seeing some odd behavior with mapped tasks and calling AWS Batch jobs. Essentially I have one mapped task that kicks off N batch jobs and returns a list of batch job ids. The next mapped task takes those job ids as input and runs

AWSClientWait

to monitor the status of each of those jobs. Here's the code for that mapped task -

@task
def wait_batch(job_id, delay, max_attempts):

    logger = prefect.context.get('logger')

    <http://logger.info|logger.info>(f"Waiting for job to complete: {job_id}")

    waiter = AWSClientWait(
        client='batch',
        waiter_name='JobComplete',
    )
    waiter.run(
        waiter_kwargs={
            'jobs': [job_id],
            'WaiterConfig': {
                'Delay': delay,
                'MaxAttempts': max_attempts
            }
        },   
    )

    <http://logger.info|logger.info>(f"Job complete: {job_id}")

    return job_id

But what we are sometimes seeing are one or more batch jobs failing, which then somehow stops the other jobs from responding to this AWSClientWait call... and so the mapped task keeps running even though all the jobs have either failed or completed. Any ideas?

I was running a prefect flow but facing a run out of memory issue, how can I clear up and free the space. I saw a 17G .prefect file in the root directory, can I remove it?

Hey! I was wondering if there's a simple way to do windowed mapping. Imagine the first task is a mapping with outputs

[1, 2, 3, 4]

. In different points I want to do overlapping and non-overlapping windows. So, 4 mapped tasks with inputs

[(1, 2), (2, 3), (3, 4)]

and then (separately) another group of 2 mapped tasks with inputs

[(1, 2), (3, 4)]

. Is this possible?

Can you specify Executor at runtime on Cloud/Server?

Hello, the Prefect Community Trivia Team will be doing our trivia round in 10 mins. DM me if you want a link to join us.

price asking. please let us know what happened.

General Q: What's the best way of dealing with flows with python package dependencies? For example, I use PySFTP, PyODBC, SqlAlchemy and pandas. So when I deploy this and it runs on an azure vm with none of these dependencies installed.. how best to deal with it in terms of... which storage? which agent? Do I package it up as a docker image as per @Kevin Kho’s excellent article here: https://medium.com/the-prefect-blog/the-simple-guide-to-productionizing-data-workflows-with-docker-31a5aae67c0a Looking for general advice / see what you guys do

#prefect-community Hey! Can we install the Prefect server on Window server 2019?

Hey guys, with the scheduling I'm just struggling for the server to actually pick it up. I want to fire my flow at 8pm each day and only on weekdays so:

flow.schedule = Schedule(
        clocks=[IntervalClock(timedelta(days=1))],
        filters=[at_time(datetime.time(20)),is_weekday]
    )
flow.schedule.next(10) #<- I have also added this line in but it's done nothing

"Couldn't connect to Prefect Server at *http://localhost:4200/graphql*" i get this when i try to run docker version in my local how to fix this anyone please help

Hello! Question about scheduling + flow dependencies. Say I have "flow a" which should run @ 4am UTC daily, "flow b" should run @ 5am UTC daily, and then "flow c" which depends on successful completion of "flow a" and "flow b" and should be run as soon as they are both complete.

Is this possible to schedule in Prefect server?

Hi, I’m trying to get a Datadog agent running in k8s to collect application metrics from my Prefect Flows. Just wondering if anybody has this working and if they could share how they did it.

Hi everyone. I am running into an issue where a task is failing due to heartbeat. I have a master flow that kicks off another flow with the

create_flow_run

task. The child flow takes over 3 hours and completed successfully. However, the parent flow's task failed with

No heartbeat detected from the remote task; marking the run as failed.

I am using Prefect Cloud. Please see thread for the code I am using to run the child flow. Thanks!

Are the extras not available with Orion? If not, how would I install the equivalent of the

google

extras?

Another question 🙂 is it possible to have the following constraint: Schedule "flow a" to run every 10 minutes & prevent scheduling "flow a" if there is an existing invocation still running.

Hello Prefect family, We currently are using prefect cloud with an ECS agent. Jobs get submitted successfully, but the time between

submitted

and

running

is 30-45 seconds. I suspect there’s some polling interval, but I don’t know where this polling interval constant value lives in the Prefect Code to lower it. Any help on this would be great! Thanks

Hello, is there a recommended approach for making a local Python module accessible to multiple docker containers? I’m running on ECS Fargate with flows stored in S3. I had hoped the serialization process would have pulled out the required objects from the import statement at build time but that doesn’t appear to be the case. Is it better practice to COPY the module in each container’s Dockerfile instead?

I need to add users to my account - I had an error “Your team has no users available. You can add more from the Account Page” - I removed a member that we no longer need - and… I still have the error that there are no users available!

Hello! I'm using the

prefect.tasks.prefect.flow_run

module to create and wait for flow runs, and have a question of how this is handled. Say I have Workflow A which (within it) submits and then waits for Workflow B. So the waiting step is a specific task within Workflow A. Is that waiting task sitting idle in the executor (e.g. is it using up a Dask worker)?

Hi. I wanted to ask you if anyone has an example of how to use `SecretBase`task. What I need to do is take a

PrefectSecret

different, depending on a value of a parameter. For example: If the param

env

is ‘dev’ i need to take

db_dev_url

PrefectSecret, otherwise takes

db_prod_url

PrefectSecret. Does it help me to do that? Or what do you think is the best way to get it? thanks

Hi, is there a way to check within my task code whether cancellation has been requested? I.e. within a potentially long operation I'd like to poll or be notified that cancellation was requested so I can raise an exception and end the task

I’ve just created a parent flow of flows and it seems like when I cancel the parent flow, the child flow that it’s running doesn’t get cancelled and continues to run. Is there a way for me to pass the cancellation down to the children flow level?

Hello! My company is new to using Prefect. I am in the process of setting up an ECS Agent and am able to have the agent successfully deploy my flow runs. However, my flows always show the below error. I’ve found some resources online that suggested to ensure that both the prefect and python version of the flow register and agent are the same and I’ve confirmed that they are. Has anyone ever experienced this before? Thanks in advance.

@Kevin Kho @Anna Geller how do I use Prefect’s

.map()

function when using the imperative prefect api? (The docs only show examples with the functional api ) I’ve tried using the

mapped

argument but that doesn’t seem to be right 🤔:

flow.set_dependencies(
            upstream_tasks=task_that_returns_a_dictionary,
            task=task_that_should_do_something_for_a_single_element,
            mapped=True,
            keyword_tasks=dict(single_element=task_that_returns_a_dictionary)
        )

This is the error message:

TypeError: Task is not iterable. If your task returns multiple results, pass `nout` to the task decorator/constructor, or provide a `Tuple` return-type annotation to your task.

Is there a way to automatically prune old images from a prefect docker agent to free up space on the box running my prefect flows?