all_finished

ValueError: Flow.run received the following unexpected parameters: unused_param1, unused_param5

import IPython
from prefect import task, Flow, Parameter


my_list = [1, 3, 4]


@task
def map1_fn(item):
    new_item = [item+1, item+1]
    print("map1_fn: {}".format(item))
    return new_item


@task
def map2_fn(item):
    # expecting item = 2 but got [2, 2] etc
    print("map2_fn: {}".format(item))
    return item


with Flow("workflow") as flow:
    my_other_list = map1_fn.map(item=my_list)
    my_another_list = map2_fn.map(item=my_other_list)

flow.run()

add_flow

python_dependencies

task retries

retry

if loop_count == 40:

from prefect import Flow, task, context
from prefect.engine.signals import LOOP
import datetime
import time

@task(max_retries=50, retry_delay=datetime.timedelta(seconds=30))
def test_retry():
    loop_count = context.get("task_loop_count", {})
    print(loop_count)

    if loop_count == 40:
        return "finished"

    raise LOOP(message="Next run")

with Flow("test-pipeline") as flow:
    test_retry()

flow.run()

Executors

DaskExecutor

mapping

click

def add(x, y):
    return x + y

def add_to_x(y):
    x = Parameter('x')
    total = task(add)(x, y)

with Flow('flow') as f:
    add_to_x(1)
    add_to_x(2)

x

get_module_metadata

dask-worker[65048]: distributed.core - INFO - Event loop was unresponsive in Worker for 3.21s.  This is often caused by long-running GIL-holding functions or moving large chunks of data. This can cause timeouts and instability.

flow

flow

main()

flow

main()

flow