Prefectとは

AirflowやDigDagと同じワークフローエンジンです。

このブログでも紹介していますが、

などがわかりやすいと思います。

英語でもよければ、公式ページ（core部分とオーケストレーションの二つ）がわかりやすいです。

Prefect Orionとは

2021/10/6にPrefectがアナウンスした「our second-generation workflow engine」です。

Say 👋 to Orion, our second-generation workflow engine!

It's DAG-free, has an observable rules engine, and delivers an incredible developer experience from test to production.

You might just love your workflows again. https://t.co/kH7bd1MLF3
— Prefect (@PrefectIO) 2021年10月5日

まだTechnical Previewで出ている情報も少ないですが、Prefect Orionについて調べてみようというのが、今回の記事です。

正直、まだぼんやりしています。

なにはともあれ触ってみる

Getting startedがあります。

下例では既存のPrefectと分けたいのでPoetry使っていますが、適当に読み替えてください。

# インストール
poetry new advent_calendar_prefect
cd advent_calendar_prefect/advent_calendar_prefect
poetry add "prefect>=2.0.0a"
poetry shell

basic.pyとして保存します。

from prefect import flow, task
from typing import List
import httpx


@task(retries=3)
def get_stars(repo: str):
    url = f"https://api.github.com/repos/{repo}"
    count = httpx.get(url).json()["stargazers_count"]
    print(f"{repo} has {count} stars!")


@flow(name="Github Stars")
def github_stars(repos: List[str]):
    for repo in repos:
        get_stars(repo)


# run the flow!
github_stars(["PrefectHQ/Prefect", "PrefectHQ/miter-design"])

実行

python basic2.py  
07:48:04.506 | Beginning flow run 'merciful-unicorn' for flow 'Github Stars'...
07:48:04.506 | Starting executor `SequentialExecutor`...
07:48:04.731 | Submitting task run 'get_stars-e40861f0-0' to executor...
PrefectHQ/Prefect has 7906 stars!
07:48:05.229 | Task run 'get_stars-e40861f0-0' finished in state Completed(message=None, type=COMPLETED)
07:48:05.319 | Submitting task run 'get_stars-e40861f0-1' to executor...
PrefectHQ/miter-design has 12 stars!
07:48:05.796 | Task run 'get_stars-e40861f0-1' finished in state Completed(message=None, type=COMPLETED)
07:48:05.800 | Shutting down executor `SequentialExecutor`...
07:48:05.882 | Flow run 'merciful-unicorn' finished in state Completed(message='All states completed.', type=COMPLETED)

サーバを起動

prefect orion start

f:id:toukoudo:20211210075028p:plain — prefet_orion

サーバを起動する前に実行したFlowも登録されていることに注目です。

（~/.prefect/orion.dbに情報が保存されるようです）

エラー

常に起きるかは不明ですが、自分の場合以下のエラーがでました（抜粋）。

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: flow
[SQL: INSERT INTO flow (id, created, updated, name, tags) VALUES (?, ?, ?, ?, ?) ON CONFLICT (name) DO NOTHING]
[parameters: ('cb63d353-0279-4e4c-adfd-5bb1254b88d7', '2021-12-06 22:33:35.319780', '2021-12-06 22:33:35.319804', 'my-favorite-function', '[]')]
(Background on this error at: https://sqlalche.me/e/14/e3q8)

DBをリセットするコマンドを実行すると動くようになりました。