@Marvin Task with a for loop inside is repeatedly executing the task flooding the logs

It appears that there are no open issues directly matching your problem on the GitHub repository for Prefect. However, it's possible that the repeating task behavior you're observing is due to a common mistake related to the use of loops within Prefect tasks. Typically, if you're using a for loop inside a Prefect task and you're experiencing logs being flooded with repeated executions, this might indicate that something within your loop is unintentionally triggering the task to be executed multiple times, such as having a

start

signal within the loop that shouldn't be there or lacking proper exit conditions. If you provide me with more context or a code snippet, I might be able to assist you further in debugging this issue. Meanwhile, I suggest you to check the Prefect documentation on loops and ensuring your tasks are idempotent, which means they can be run multiple times without unexpected side effects. If you need examples of how to properly implement loops in a Prefect flow, I can fetch some code examples for you.

@Marvin can you fetch an example for me?

Here is an example of a Prefect flow without an explicit loop but demonstrates chaining task executions, which may help you understand how to manage task execution without flooding the logs:

python
from prefect import flow, task

@task(task_run_name="Adding {x} and {y}")
def add(x: int, y: int) -> int:
    """Just a python function with a fancy hat on!"""
    return x + y

@flow
def my_flow():
    """Tasks must be called in the flow to be executed (at this time)."""
    first_result = add(1, 2)
    second_result = add(first_result, first_result)
    return add(first_result, second_result)

if __name__ == "__main__":
    # Run the flow
    my_flow()
    
    # You can run tasks' functions directly outside of a flow if you want
    # but it won't be tracked by Prefect, it's just a normal function
    assert add.fn(1, 2) == 3

(Link to code example source: GitHub - Basic Task Example) In this example, each task is called linearly within the flow. If you're looking to implement loops, ensure that you have control over the conditions on which tasks are re-executed to prevent unintended re-runs. If you require a loop, you would typically use a for-loop around your task invocations within the flow function, being cautious of whether the tasks within the loop should be awaited or run concurrently. Remember that each invocation of a task function within a Prefect flow corresponds to a task run that will be logged. Would you like a more explicit example featuring a loop within a Prefect flow?

@Marvin is it possible to see what task is next in queue in a flow?

I'm afraid that there's no direct feature in Prefect that explicitly shows what task is next in the queue for a flow run as if you were monitoring a typical queue structure. However, you can infer the order of task execution by the flow definition and the dependencies between tasks. If you define your flow and tasks with explicit dependencies—where one task must be completed before the next one starts—the logs produced during a flow run can give you an idea of the order in which tasks are being executed. Here is an excerpt showing how the dependencies can determine the task execution order:

@flow(log_prints=True)
def flow_of_tasks():
    # with dependencies, tasks execute in order
    first_result = first.submit()
    second_result = second.submit(first_result)
    third.submit(second_result)

@task
def first():
    print("I'm first!")

@task
def second(input):
    print("I'm second!")

@task
def third(input):
    print("I'm third!")

In the flow logs, you will see the creation and submission of task runs in order, followed by their completion:

Flow run 'pumpkin-puffin' - Submitted task run 'first-0' for execution.
Task run 'first-0' - I'm first!
Task run 'first-0' - Finished in state Completed()
Flow run 'pumpkin-puffin' - Created task run 'second-0' for task 'second'
... (and so on)

For more information, you can refer to the Prefect documentation on specifying upstream dependencies: - Successfully Deploying Task Queue - Specifying Upstream Dependencies To actively view or predict which task is going to run next, you might need to consider integrating an external logging or monitoring solution with more granular control over task queuing visibility.