Prefect（Python）とは？ワークフロー管理の使い方と3系の始め方

Prefectは、Pythonの関数を@flow・@taskで囲むだけでデータパイプラインを構築・監視できるワークフローオーケストレーションツールです（英単語の prefect＝監督生・長官の意味ではなく、開発元 Prefect Technologies が提供するOSSツールを指します）。2024年9月にメジャー更新の3.0がGAとなり、本稿執筆時点の最新は3.7.8です。この記事では、Prefectの基本概念、pip導入とDocker Composeでの環境構築、FlowとTaskの書き方、デプロイとWork Pool、スケジューリング、キャッシュによる冪等性、AirflowやDagsterとの使い分けまで、最新の3系にそろえて解説します。

まとめ：Prefectの要点と最短の始め方

• 正体：Pythonネイティブのワークフローオーケストレーションツール。既存のPython関数にデコレータを付けるだけでフロー化でき、実行状況をUIで可視化・リトライ・スケジュール実行できる。
• 3系の要点：イベント駆動オートメーションがOSSに開放、トランザクションAPIで冪等性を確保、実行基盤はエージェント廃止でWork Pool＋Workerへ統一、実行エンジン刷新で高速化。
• 最短の始め方：pip install prefect → 関数に@task/@flow → python flow.py で実行。UIはprefect server startでhttp://127.0.0.1:4200。
• 使いどころ：Pythonで動的にパイプラインを書きたいチーム向き。静的DAGの大規模バッチを既にAirflowで回しているなら無理に移行しない、長時間の分散ステートフル処理はTemporalが候補、という線引きで選ぶ。

Prefectとは｜Pythonで書くワークフローオーケストレーションツール

Prefectは、複数の処理（データ抽出→変換→ロード、モデル学習、バッチ集計など）を「いつ・どの順で・失敗したらどうするか」を含めて管理する仕組みを、Pythonコードのまま与えます。ライセンスはApache 2.0のOSS本体に加え、マネージド版のPrefect Cloudがあり、同じコードをそのまま自前サーバーでもCloudでも動かせます。特別なDSLやYAMLでDAGを宣言する必要はなく、通常のPython関数にデコレータを付けるだけで、リトライ・キャッシュ・ログ・可視化が後付けで効きます。

まず押さえるべき構成要素は次の5つです。この語彙が分かると、以降のインストールやデプロイの説明が一直線に理解できます。

同種のツールにApache AirflowやDagster、Temporalがあります。Airflowが静的なDAG定義と豊富な運用実績で強いのに対し、Prefectは「普通のPythonをそのままワークフロー化する」動的な書き味が特徴です。比較の詳細は本文後半のApache Airflowによるワークフロー管理との違いの章で扱います。

Prefect 3系の主な特徴と2系からの変更点

3.0は2.xからの単なる機能追加ではなく、実行エンジンとキャッシュ、実行基盤を作り替えたメジャー更新です。2系の記事やコードを見て学ぶ場合は、次の変更点を前提に読み替えてください。

最大の変更は、これまでPrefect Cloud専用だったイベント＆オートメーション基盤がOSS本体に入ったことです。フローの完了や外部イベントをトリガーに別のフローを起動する、といった自動化を無料のOSSだけで組めます。

もう一つの核がトランザクションAPIです。複数のTaskを1つのトランザクションにまとめ、途中で失敗したら副作用（書き込み済みファイルの削除など）をロールバックできます。これによりキャッシュも「トランザクションがコミットされた時だけ結果を確定する」方式に刷新され、再実行時の冪等性が扱いやすくなりました。

実行基盤ではエージェントが廃止され、Work Pool＋Workerに一本化されました。2系でagentを使っていた場合、3系ではWorkerへの置き換えが必要です。あわせて、既定でコードがメインスレッド上を走るようになり、共有リソースやスレッドセーフでない処理での挙動が直感的になったうえ、実行エンジンの刷新で大規模フローの処理性能が改善しています。

Prefect 3系のインストールと環境構築

導入はPythonパッケージ1つで完結します。ローカルで試す→Docker Composeで永続化する→Cloudに載せる、の順で環境を広げていくと迷いません。

pipでのインストールとバージョン確認

Python 3.10以上の環境で次を実行します（最新3.7系はPython 3.10以上が必要）。仮想環境（venvやuv）を切ってから入れるのが安全です。

pip install prefect
prefect version

prefect versionで3.7.x系が表示されればOKです。2系が入っている環境をアップグレードする場合は、上のトランザクション表の変更点（特にエージェント廃止）を確認してから上げてください。

ローカルサーバー（UI）の起動と接続

手元でダッシュボードを立ち上げるにはサーバーを起動します。

prefect server start

起動後、ブラウザでhttp://127.0.0.1:4200を開くとUIが表示され、フローの実行履歴やログ、Work Poolの状態を確認できます。フロー実行時にこのローカルAPIへ接続させたい場合は、別ターミナルでprefect config set PREFECT_API_URL=http://127.0.0.1:4200/apiを設定します。

Docker Composeで永続構成を作る

prefect server start単体はSQLiteの一時構成で、プロセスを止めると実行履歴が扱いにくくなります。チームで使うなら、PostgreSQLをバックエンドにしたサーバーをDocker Composeで永続化するのが実務的です。Prefectサーバーは環境変数PREFECT_API_DATABASE_CONNECTION_URLにPostgreSQLの接続文字列を渡すだけでDBを切り替えられます。

services:
  database:
    image: postgres:16
    environment:
      POSTGRES_USER: prefect
      POSTGRES_PASSWORD: prefect
      POSTGRES_DB: prefect
    volumes:
      - db:/var/lib/postgresql/data
  server:
    image: prefecthq/prefect:3-latest
    command: prefect server start --host 0.0.0.0
    environment:
      PREFECT_API_DATABASE_CONNECTION_URL: postgresql+asyncpg://prefect:prefect@database:5432/prefect
    ports:
      - "4200:4200"
    depends_on:
      - database
volumes:
  db:

Docker Compose自体の書き方（サービス定義やボリューム管理）に不安があれば、docker-composeで複数コンテナを一括管理する仕組みを先に押さえておくと、この構成の意味が理解しやすくなります。

Prefect Cloudを使う場合（OSSとの違い）

サーバー運用を自前で抱えたくない場合はPrefect Cloudを使います。CLIからログインしてワークスペースに接続すれば、フローの実行結果が自動でCloud側に集約されます。

prefect cloud login

OSSとCloudでフローのコードは同一で、違いは「オーケストレーション面（履歴保存・イベント基盤・権限管理・SLA監視）を誰が運用するか」です。Cloudには無料枠があり、まず個人利用で試してからチーム利用へ広げられます。料金や上限はプランで変わるため、採用前に公式の最新プランを確認してください。

FlowとTaskの最小実装

Prefectの学習コストが低い理由は、通常のPython関数にデコレータを足すだけでワークフロー化できる点にあります。次はTaskを1つ定義し、Flowから呼び出す最小例です。

from prefect import flow, task

@task
def add(x: int, y: int) -> int:
    return x + y

@flow(log_prints=True)
def my_flow():
    total = add(3, 5)
    print(f"3 + 5 = {total}")

if __name__ == "__main__":
    my_flow()

実行はスクリプトをそのまま走らせるだけです。

python flow.py

2系の解説記事に出てくるprefect runコマンドは3系では使いません。開発中はpython flow.pyで直接実行し、スケジュールやAPI経由の実行が必要になった段階で次章のデプロイに進む、という流れになります。@flow(log_prints=True)を付けるとprint出力がそのままフローのログとしてUIに残ります。

デプロイとWork Pool・Workerによる実行

スケジュール実行やAPIからのトリガー実行をするには、Flowを「デプロイ」して常時受け付けられる状態にします。3系での基本パターンは2通りです。

手軽なのは.serve()で、指定したFlowを常駐プロセスとして待機させ、スケジュールやUIからの実行を受け付けます。

if __name__ == "__main__":
    my_flow.serve(name="my-first-deploy", cron="0 9 * * *")

本番でインフラを分けたい場合は、Work PoolとWorkerを使います。Work Poolは「どこで実行するか」の窓口、WorkerはそのPoolを監視して実際にジョブを起動するプロセスです。

prefect work-pool create my-pool --type process
prefect worker start --pool my-pool

あとはFlowをflow.deploy()またはprefect deployで対象のWork Poolに紐付けて登録すれば、スケジュールやAPI実行のたびにWorkerがジョブを起動します。ローカル検証はprocessタイプ、本番はDockerやKubernetesなどのPoolタイプ、と実行基盤に合わせて選び分けます。

スケジューリングと自動化（Automation）

定期実行は、デプロイにスケジュールを付けるだけで有効になります。CronとIntervalの2方式があり、.serve()やprefect deployの引数、またはUIから設定できます。

# 毎日9時に実行（Cron）
my_flow.serve(name="daily", cron="0 9 * * *")

# 1時間ごとに実行（Interval）
my_flow.serve(name="hourly", interval=3600)

時刻ベースの定期実行に加え、3系ではイベント駆動のAutomationがOSSで使えます。「あるフローが失敗したらSlackに通知する」「特定のイベントを受けたら別フローを起動する」といったルールを、UIまたはコードで定義できます。失敗時のリトライはTask/Flowの引数で指定します。

@task(retries=3, retry_delay_seconds=10)
def call_api():
    ...

このretriesとAutomationの通知を組み合わせると、「3回リトライしても失敗したら通知」という運用が定型化できます。通知チャネル（メール・Slack等）はブロック（Block）として登録して再利用します。

キャッシュとトランザクションによる冪等性の確保

同じ入力に対する再計算を避けたい、あるいは途中失敗した処理をきれいにやり直したい——こうした冪等性の要求に、3系はトランザクションとキャッシュポリシーで応えます。まずTaskのキャッシュはcache_policyで制御します。

from prefect import task
from prefect.cache_policies import INPUTS

@task(cache_policy=INPUTS)
def transform(data):
    ...

INPUTSを指定すると、同じ引数で呼ばれたTaskは前回結果を再利用してスキップします。さらに複数Taskをまとめて「全部成功したら確定、途中で失敗したら巻き戻す」制御をしたい場合はトランザクションを使います。

from prefect import flow
from prefect.transactions import transaction

@flow
def pipeline():
    with transaction():
        write_file()   # 失敗時はon_rollbackで削除
        load_db()

トランザクション内のTaskにon_rollbackフックを定義しておくと、後段のTaskがこけた時に前段の副作用（書き込んだファイルの削除など）を自動で取り消せます。バッチ処理の「途中まで書き込まれて中途半端な状態が残る」問題を、コードの構造として防げるのが3系の強みです。

クラウド・インフラ連携（Docker / Kubernetes / ECS / Cloud Run / GCP）

本番ではフローをコンテナやマネージド基盤で動かします。実行先はWork Poolのタイプで切り替えるだけで、Flowのコードは変えません。代表的なタイプは次の通りです。

KubernetesでのJob実行は、クラスタ側の権限やイメージ管理を含めた設計が要ります。前提となるK8sの挙動はKubernetes 1.34の新機能・変更点の解説を、AWSでのコンテナ実行はAWS ECSの仕組みと起動タイプおよびAWS Fargateの料金とEC2との違いをあわせて確認すると、どのWork Poolを選ぶかの判断が付きます。GCP連携はprefect-gcp、AWS連携はprefect-awsのように、統合パッケージを追加インストールして使います。

pip install prefect-gcp   # Cloud Run・Vertex AI 連携
pip install prefect-aws   # ECS・S3 連携

PrefectとAirflow・Dagsterの違いと、採用すべきでない場面

「結局どれを使うか」で迷うなら、書き味と運用モデルで割り切るのが早いです。主要ツールの性格を並べます。

Prefectを積極的に選ぶのは、Pythonチームが動的なパイプライン（実行時に分岐やループ数が変わる処理）を素早く書きたいときです。逆に、次のような場面ではPrefectを第一候補にしない方が無難です。

• 既にAirflowで安定運用している静的な大規模バッチ：移行コストに見合うだけの利点が出にくい。まずAirflowのまま最適化する。Apache Airflowによるワークフロー管理で自社の運用と照らして判断する。
• 数時間〜数日にわたる長時間・分散のステートフル処理：耐久性の高いワークフロー実行が本質ならTemporal Workflowの仕組みと比較が適する領域。
• データ資産のリネージ・品質管理が主目的：アセット中心のDagsterの方がモデルに合う。

Prefectは「オーケストレーションを重装備で導入する」より「今あるPythonコードに信頼性を後付けする」方向に強いツールだ、と位置づけると選定を誤りません。

よくある質問（FAQ）

prefectは英単語の「監督生」とは別物ですか？

別物です。英単語の prefect は監督生・長官を指しますが、本稿のPrefectは Prefect Technologies が開発するPython製のワークフローオーケストレーションツールを指します。検索意図が辞書か製品かで結果が混在するため、ツールを探す場合は「prefect python」「prefect オーケストレーション」で絞ると精度が上がります。

Prefect Cloudは無料で使えますか？

無料枠があり、個人利用や小規模な検証はコストをかけずに始められます。OSS本体（Apache 2.0）はサーバーごと無料で自前運用でき、Cloudは履歴保存・イベント基盤・権限管理などのオーケストレーション面をマネージドで肩代わりします。上限や有料プランの条件は変動するため、採用前に公式の最新プランを確認してください。

Prefect 2から3への移行は必要ですか？互換性は？

2系は保守フェーズで、新機能や性能改善は3系が対象です。3系ではエージェントが廃止されWorkerへ移行が必要、キャッシュがトランザクション方式に、型検証がPydantic v2に変わっています。既存の2系フロー自体はデコレータの書き方が近く移行しやすい一方、実行基盤（agent→worker）とキャッシュ設定は書き換えが要ります。

PrefectのGUI（UI）はどこで確認しますか？

ローカルではprefect server startを実行しhttp://127.0.0.1:4200を開くとUIが表示されます。フローの実行履歴・状態・ログ、Work PoolやWorkerの稼働状況、スケジュールの管理をこの画面で行います。Prefect Cloudを使う場合はCloudのワークスペース画面が同じ役割を担います。

AirflowとPrefectはどちらを選ぶべきですか？

静的なDAGを明示定義して大規模バッチを堅く運用したいならAirflow、通常のPythonをそのまま動的にワークフロー化して素早く回したいならPrefectが向きます。既にAirflowで安定している場合は無理に乗り換えず、新規のPythonパイプラインからPrefectを試すのが現実的です。

関連記事

• Apache Airflowを活用したビッグデータのワークフロー管理
• Temporal Workflowとは？コードで書く分散ワークフローエンジンの仕組みと比較
• docker-composeとは？複数コンテナをymlで定義し一括管理する仕組みを解説
• AWS ECSとは？コンテナの仕組み・起動タイプ・料金・EKSとの違い
• Kubernetes 1.34の新機能・変更点とEOL｜DRA GA・再起動ルールを解説