Prefectで「死なない」データパイプラインを構築する：Pythonスクリプトをサイレントエラーで終わらせないために – ITFROMZERO

Table of Contents

深夜にPythonスクリプトが「落ちる」不安におさらば

想像してみてください。仮想通貨の価格スクレイピングスクリプトを完成させ、夜の11時にCronジョブを使ってVPSにセットしたとします。あなたは安心して眠りにつき、翌朝にはデータが詰まったCSVファイルができていることを期待しています。しかし現実は非情です。11時05分にAPI接続エラーが発生し、スクリプトが停止。貴重な8時間分のデータを失ってしまいました。

以前の私は、数百MBの巨大なログファイルを何時間もかけて漁ったり、ネットワークエラーをキャッチするためだけに大量の try...except ブロックを書いたりすることに時間を費やしていました。しかし、Prefect に切り替えてからすべてが変わりました。このツールは、断片的なコードを、自己修復能力と視覚的な監視機能を備えた堅牢なパイプラインシステムへと変えてくれます。

クイックスタート：2分で最初のパイプラインを実行する

Prefectを導入するために、ゼロから学習し直す必要はありません。使い慣れた純粋なPythonコードをベースに動作します。まずはライブラリをインストールしましょう。

pip install -U prefect

普通の天気データ取得スクリプトを、@task と @flow デコレータを使ってプロフェッショナルなフローに変える方法を見てみましょう：

from prefect import task, flow
import random

@task(retries=3, retry_delay_seconds=10)
def get_data():
    # 30%の確率でAPIエラーをシミュレート
    if random.random() > 0.7:
        raise ValueError("APIが応答しません！")
    return [25, 28, 30, 22]

@task
def transform_data(data):
    return [x * 1.8 + 32 for x in data] # 摂氏から華氏に変換

@flow(name="Weather Pipeline")
def weather_flow():
    raw = get_data()
    processed = transform_data(raw)
    print(f"温度 (F): {processed}")

if __name__ == "__main__":
    weather_flow()

ここでの最大のメリットは retries=3 です。APIに問題が発生した場合、Prefectは10秒後に自動的に再試行します。複雑なリトライロジックを自分で書く必要はありません。すべての実行プロセスは、ターミナル上で詳細にログ記録されます。

なぜPrefectはCronジョブやAirflowよりも優れているのか？

「軽量なCronで十分ではないか？」「エンタープライズ標準のAirflowを使うべきでは？」と疑問に思うかもしれません。実戦経験から得た、Prefectを選ぶべき3つの核心的な理由は以下の通りです。

1. 非常に優れた観測性（Observability）ダッシュボード

Cronジョブでは、スクリプトの状態が全く把握できません。Prefectなら、prefect server start と入力するだけで、localhost:4200 でダッシュボードが立ち上がります。ここでは、実行時間のチャート、リソースを最も消費しているタスク、そしてタスクが失敗した際の正確な原因を確認できます。

2. エクスポネンシャルバックオフによるスマートなエラー処理

約50万件の顧客レコードを処理する場合、データベースの過負荷は日常茶飯事です。Prefectは `exponential backoff` をサポートしており、リトライ間の待ち時間を段階的に増やすことができます。これにより、サーバーがダウンしている時にリクエストを「爆撃」して追い打ちをかけることを防ぎます。

3. コードファースト：通常のPythonを書く感覚で実装可能

Airflowでは、DAGという制約の多い、やや冗長な構造でコードを書く必要があります。Prefectは違います。既存のロジックはそのままに、デコレータでラップするだけです。開発者の書き方を尊重しているため、古いスクリプトから新しいパイプラインへの移行は数分で完了します。

アドバンス：キャッシュによるリソースの節約

例えば、Google Driveから2GBのレポートファイルをダウンロードする必要があるとします。後の処理ステップでエラーが出たからといって、このファイルを何度も再ダウンロードしたくはないでしょう。Prefectは cache_key_fn でこの問題を解決します：

from prefect.tasks import task_input_hash
from datetime import timedelta

@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=2))
def download_heavy_file(file_id):
    # file_idが2時間以内に変更されない場合、このタスクは再実行されません
    print("非常に重いファイルをダウンロード中...")
    return "データ内容"

サーバーへのデプロイも一瞬で完了

毎日午前8時にパイプラインを自動実行するために、Linuxの crontab -e を触る必要はありません。deployコマンドを使用しましょう：

prefect deploy weather_script.py:weather_flow -n "Daily-Check" --cron "0 8 * * *"

このスケジュールは、Webインターフェース上で簡単にオン/オフを切り替えたり編集したりできます。サーバーにSSHで入ることなく実行時間を変更できるのは、非常に便利です。

大規模プロジェクトからの実践的な知見

1年以上PrefectでETLシステムを管理してきた中で得た、4つの重要な教訓を紹介します：

タスクを最大限に細分化する： データの取得、処理、DB保存を1つのタスクにまとめないでください。DB保存でエラーが起きた場合、コストのかかるデータ取得からやり直す羽目になります。
セキュリティのためにBlocksを使用する： APIキーを .env ファイルに置いて漏洩のリスクに晒すのではなく、Prefect Blocksに保存しましょう。コードがよりクリーンで安全になります。
管理のためにタグを付ける： パイプラインが20を超えたら、production や crawling といったタグを付けて、ダッシュボードですぐにフィルタリングできるようにしましょう。
リトライを乱用しない： ゼロ除算のようなロジック上のエラーの場合、100回リトライしても解決しません。リトライはネットワークやタイムアウトなどの外部要因によるエラーにのみ使用してください。

データパイプラインを構築すること自体は難しくありませんが、それを安定して稼働させ続けることこそが真の挑戦です。Prefectは複雑なインフラ部分を肩代わりしてくれるため、あなたはデータ処理のロジックに集中できます。バックグラウンドスクリプトの管理に疲れているなら、今すぐPrefectを試してみてください。