Prefect 플로(Flow)

Prefect 플로(Flow)

Prefect에서 플로(flow)는 워크플로의 기본 단위예요. 파이썬 함수에 @flow 데코레이터를 붙이면 그 함수가 곧 플로가 되고, 플로 안에서 태스크들을 호출해 전체 파이프라인을 구성할 수 있어요. 일반 파이썬 함수처럼 호출해서 실행할 수 있을 만큼 진입장벽이 낮아요.

출처: Prefect Flows

플로 정의

가장 단순한 플로는 이렇게 만들어요.

from prefect import flow

@flow
def my_workflow() -> str:
    return "Hello, world!"

실행은 보통 파이썬 함수를 호출하듯 하면 돼요.

my_workflow()

플로는 일반 함수뿐 아니라 클래스 메서드·제너레이터 함수에도 적용할 수 있어요. 예를 들어 제너레이터 플로는 값을 하나씩 yield 하면서 실행을 이어나갈 수 있죠. 실행된 플로 런은 Prefect UI에서 확인할 수 있어요.

태스크와 자식 플로

플로 안에서 @task 데코레이터를 붙인 태스크를 호출하고, 심지어 다른 플로를 자식 플로로 호출할 수도 있어요.

from prefect import flow, task

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

@flow
def is_number_even(number: int):
    return number % 2 == 0

@flow
def even_or_odd():
    number = generate_a_number()
    if is_number_even(number):
        ...

이렇게 하면 실행 단위를 태스크와 플로 두 수준으로 나눠 관리할 수 있어요.

타임아웃과 재시도

플로가 오래 돌거나 실패했을 때의 정책도 지정할 수 있어요. timeout_seconds 키워드로 실행 제한 시간을 두면 너무 오래 도는 플로를 막을 수 있어요.

@flow(timeout_seconds=1, log_prints=True)
def show_timeouts():
    print("I will execute")
    time.sleep(5)
    print("I will not execute")

실패 시 재시도 횟수는 retries 인자로 조정할 수 있고, version으로 플로 버전을 명시할 수도 있어요.

플로 설정 인자

@flow 데코레이터는 여러 설정 인자를 받아요. 예를 들어 description(설명), retries(재시도 횟수), version(버전 문자열) 등이 있어요. description을 생략하면 데코레이트된 함수의 docstring을 자동으로 가져와요.

더 알아보기