데이터 수집 스크립트에 재시도와 로그를 붙이다 보면 업무 로직보다 실행 관리 코드가 커지기 쉽다. Prefect는 Python 함수에 실행 상태, 재시도, 배포와 관찰 기능을 더하는 오케스트레이터다.
이 글은 Prefect 3의 Flow·Task와 배포 방식을 기준으로 한다. Prefect 2 시기의 agent 중심 설명과 구분하고, 로컬 실행에서 worker 기반 운영으로 이어지는 경계를 정리한다.
1. Flow는 작업 흐름, Task는 관찰할 실행 단위
@flow는 전체 작업 흐름을, @task는 개별 실행 단위를 정의한다. 일반 Python의 조건문과 반복문을 사용할 수 있으므로 기존 스크립트를 단계적으로 옮기기 쉽다.
모든 함수를 task로 만들 필요는 없다. 별도로 재시도하거나 상태·실행 시간을 관찰할 이유가 있는 지점을 기준으로 나눈다. 입력 변환 보조 함수까지 지나치게 잘게 나누면 관리 비용이 커질 수 있다.
| 구분 | 역할 |
|---|---|
| Flow 정의 | 작업 흐름과 매개변수 정의 |
| Flow Run | 특정 입력으로 시작된 한 번의 실행 |
| Task Run | Flow 안에서 수행되는 작업의 실행 기록 |
| Deployment | 실행할 코드·매개변수·스케줄·인프라 설정 연결 |
2. 고정 입력으로 먼저 실행해 보기
외부 API의 응답에 의존하지 않는 작은 예제부터 시작한다. 아래 코드는 날짜별 금액을 합산하며, 결과는 {'2025-01-01': 250}이다. Python 3.12와 Prefect 3을 사용하는 별도 가상환경에서 실행할 수 있다.
python -m pip install "prefect>=3.4,<4"
python sales_flow.py# sales_flow.py
from prefect import flow, task
@task(retries=2, retry_delay_seconds=1, persist_result=False)
def extract() -> list[dict]:
return [
{"date": "2025-01-01", "amount": 100},
{"date": "2025-01-01", "amount": 150},
]
@task(persist_result=False)
def aggregate(rows: list[dict]) -> dict[str, int]:
totals: dict[str, int] = {}
for row in rows:
amount = row["amount"]
if type(amount) is not int or amount < 0:
raise ValueError("amount must be a non-negative integer")
day = row["date"]
totals[day] = totals.get(day, 0) + amount
return totals
@flow(log_prints=True, persist_result=False)
def sales_flow() -> dict[str, int]:
summary = aggregate(extract())
print(summary)
return summary
if __name__ == "__main__":
sales_flow()이 예제는 순차 실행이다. task로 감쌌다는 이유로 동시에 실행되는 것은 아니다. 병렬 실행이 필요하면 .submit()·.map()과 task runner, 외부 시스템의 동시 요청 제한을 함께 검토한다.
로컬 실행은 Prefect의 실행 상태 관리도 사용한다. 서버 연결 설정이 없다면 임시 로컬 서버가 사용될 수 있으므로 네트워크·파일 권한과 실행 로그를 확인한다. 업무 함수만 별도로 시험하려면 task의 .fn으로 순수 계산 부분을 호출할 수 있다.
3. 재시도할 오류를 구분하기
retries=2는 처음 실행한 뒤 최대 두 번 더 시도하는 설정이다. 예제에서는 향후 외부 수집으로 바뀔 수 있는 extract에 재시도를 두고, 입력 형식이 잘못된 집계 오류에는 무조건 재시도하지 않는다.
일시적인 네트워크 오류, 요청 제한, 잘못된 인증 정보, 잘못된 데이터는 대응이 다르다. 지수형 대기나 jitter가 필요한지, 어떤 예외를 재시도할지, 총 대기 시간을 얼마나 허용할지 정한다.
재시도는 외부 부수 효과를 되돌리지 않는다. 예를 들어 결과 저장 뒤 응답이 유실되면 같은 task가 다시 실행될 수 있다. 업무 날짜·원본 이벤트 ID 같은 키로 결과를 upsert하거나 중복을 확인하는 멱등성이 필요하다.
task 결과 캐시, 결과의 영속 저장, 외부 DB의 저장 성공도 다른 개념이다. persist_result=False인 위 예제는 결과 저장소를 구성한 운영 예제가 아니다. 재시작 이후에도 결과를 재사용하려면 저장소, 직렬화, 캐시 키와 만료 정책을 함께 설계해야 한다.
4. 로컬 실행과 Deployment를 구분하기
python sales_flow.py는 지금 한 번 실행한다. 정기 실행과 운영 관리가 필요하면 Deployment를 만든다. Prefect 3에서는 다음 두 방식부터 구분하면 이해하기 쉽다.
같은 환경에서 실행을 받는 serve
# serve_sales.py
from sales_flow import sales_flow
if __name__ == "__main__":
sales_flow.serve(name="sales-local")선택한 Prefect 서버 또는 Cloud에 연결한 상태에서 실행한다. serve는 Deployment를 만들고 실행 요청을 기다리는 장기 실행 프로세스다. 파일을 실행한 뒤 종료해 버리면 계속 작업을 받아 처리할 수 없다. 위 코드는 자동 스케줄을 추가하지 않은 예시이며, 운영 주기가 필요하면 시간대를 포함한 스케줄을 설정한다.
work pool과 worker로 실행 인프라 연결
work pool은 실행 인프라의 유형과 기본 설정을 관리한다. worker는 대응하는 pool에서 실행을 받아 process, Docker, Kubernetes 등 해당 환경에 작업을 제출한다. 필요한 integration 패키지와 인프라 권한도 준비해야 한다.
Prefect 3에서 agent 대신 worker를 사용하는 전환은 명령 이름만 바꾸는 작업이 아니다. Deployment의 코드 전달 방식과 infrastructure block에서 work pool로 옮겨 갈 설정을 함께 확인한다.
코드를 실행 환경에 제공하는 방법도 정해야 한다. 로컬 파일이 worker 머신에도 있는지, Git에서 어떤 버전을 받는지, 이미지에 어떤 의존성을 넣었는지가 중요하다. Deployment 등록 성공이 실행 환경의 준비 완료를 뜻하지는 않는다.
push·managed work pool처럼 별도 worker를 직접 운영하지 않는 방식도 있으므로 모든 배포를 “사용자 서버의 agent가 실행한다”라고 설명하면 부정확하다.
5. 제어면과 데이터 처리 위치를 함께 확인하기
Prefect API와 UI는 실행 상태, 스케줄과 운영 정보를 관리한다. 실제 함수는 선택한 실행 환경에서 수행된다. 이 구분은 유용하지만 “Cloud에는 코드나 데이터가 절대 전달되지 않는다”는 보장은 아니다.
매개변수, 로그, 이벤트와 결과 저장 설정에 따라 데이터가 외부로 전달되거나 저장될 수 있다. Cloud 관리형 실행을 선택했다면 코드 실행 위치 자체도 달라진다. 어떤 정보를 어디에 기록하는지 배포 방식별로 확인한다.
인증 정보는 로그에 출력하지 않고 적절한 Secret 관리 방식으로 전달한다. 실패 원인 추적에 필요한 식별자는 남기되 원본 개인정보를 통째로 기록하지 않는다. 민감한 결과라면 저장 위치와 접근 권한, 보존 기간까지 정한다.
6. 운영에서 확인할 것
| 상황 | 확인할 기준 |
|---|---|
| worker 중단 | heartbeat와 실행 대기 적체를 발견하는가? |
| 외부 API 지연 | 타임아웃·재시도가 무한 대기로 이어지지 않는가? |
| 같은 업무의 재실행 | 저장·알림 같은 부수 효과가 중복되지 않는가? |
| 입력 데이터 오류 | 재시도보다 수정이 필요한 실패를 구분하는가? |
| 코드·패키지 변경 | 실행별로 사용한 버전을 재현할 수 있는가? |
Python 제어 흐름을 자연스럽게 유지하고 기존 스크립트에 실행 관리 기능을 붙이는 것이 중요하다면 Prefect를 검토할 만하다. 다만 실행 관리 도구를 도입한다고 업무 데이터의 품질이나 최종 저장의 정확성까지 자동으로 해결되지는 않는다.
Airflow도 동적 작업 구성을 지원하므로 “정적 DAG 대 동적 Python”이라는 구분만으로 우열을 판단하지 않는다. 실행·배포 모델, 과거 데이터 재처리, 팀의 운영 경험을 같은 기준에서 비교한다.
Prefect를 사용할 때의 핵심은 작업 코드, 실행 인프라, 상태·결과 기록을 구분하고 실패 이후의 행동을 명시하는 것이다.
