How to avoid this unwated task in logs and taskrun...
# prefect-community
m
How to avoid this unwated task in logs and taskruns (in GUI as well):
Copy code
[2022-01-19 14:45:49+0530] INFO - prefect.TaskRunner | Task 'Tuple': Finished task run for task with final state: 'Success'
[2022-01-19 14:45:49+0530] INFO - prefect.TaskRunner | Task 'Tuple': Finished task run for task with final state: 'Success'
[2022-01-19 14:45:49+0530] INFO - prefect.TaskRunner | Task 'Tuple': Finished task run for task with final state: 'Success'
[2022-01-19 14:45:49+0530] INFO - prefect.TaskRunner | Task 'Tuple': Finished task run for task with final state: 'Success'
[2022-01-19 14:45:49+0530] INFO - prefect.TaskRunner | Task 'Tuple': Finished task run for task with final state: 'Success'
[2022-01-19 14:45:49+0530] INFO - prefect.TaskRunner | Task 'Tuple': Finished task run for task with final state: 'Success'
[2022-01-19 14:45:49+0530] INFO - prefect.TaskRunner | Task 'Tuple': Finished task run for task with final state: 'Success'
[2022-01-19 14:45:49+0530] INFO - prefect.TaskRunner | Task 'Tuple': Finished task run for task with final state: 'Success'
[2022-01-19 14:45:49+0530] INFO - prefect.TaskRunner | Task 'List': Starting task run...
[2022-01-19 14:45:49+0530] INFO - prefect.TaskRunner | Task 'List': Starting task run...
[2022-01-19 14:45:50+0530] INFO - prefect.TaskRunner | Task 'List': Starting task run...
[2022-01-19 14:45:50+0530] INFO - prefect.TaskRunner | Task 'List': Starting task run...
a
Anytime a non-Task input is provided to a task, Prefect automatically converts it to a special task representing the collection type e.g. List, Tuple, Dict. If you want to avoid it, you could wrap it into another task and pass it as data dependency.
Copy code
import random
from prefect import Flow, task


@task
def a_number():
    return random.randint(0, 100)


@task
def get_sum(x):
    return sum(x)


with Flow("Using Collections") as flow:
    a = a_number()
    b = a_number()
    s = get_sum([a, b])


if __name__ == "__main__":
    flow.visualize()
Compare it to this:
Copy code
import random
from prefect import Flow, task


@task
def a_number():
    return random.randint(0, 100)


@task
def merge_into_list(x, y):
    return [x, y]


@task
def get_sum(x):
    return sum(x)


with Flow("Using Collections") as flow:
    a = a_number()
    b = a_number()
    a_list = merge_into_list(a, b)
    s = get_sum(a_list)


if __name__ == "__main__":
    flow.visualize()
m
thanks !!