肉球でキーボード

MLエンジニアの技術ブログです

PrefectのFlow実行名を変更する


PrefectでFlow実行名を変更する方法を紹介します。
本文中コード: code-for-blogpost/prefect_rename_flow at main · nsakki55/code-for-blogpost · GitHub

flow実行名の変更について

PrefectのFlow実行名は、デフォルトではランダムに名前が割り当てられます。

Flowの実行ログを確認する時、分かりにくいため、意味のあるFlow実行名に変更したくなります。

PrefectではFlow実行名を変更する方法として、Flowを定義する際に@flowデコレータに flow_run_name 引数を渡す方法が提供されています。

https://docs.prefect.io/2.10.18/concepts/flows/#flow-settings

例えば、以下のように実行パラメータに時刻を渡すことで、Flow名に実行時刻を設定することができます。

from datetime import datetime

from prefect import flow, get_run_logger

@flow(flow_run_name="my-flow-{date}")
def my_flow(date: datetime = None):
    logger = get_run_logger()
    logger.info(f"date: {date}")

if __name__ == "__main__":
    now = datetime.now()
    my_flow(date=now)

flow_run_name でFlow実行名を設定する方法だと、スケジュール実行時にデフォルトの実行パラメータが渡されるため、実行時間を渡すことができない問題があります。

Flow実行名を変えるために実行パラメータを設定する必要があるため、余分な実行パラメータを増やすことにもなります。

flow_run_name を使用せずに、任意のFlow実行名を設定する方法を紹介します。

任意のflow実行名に変更

実行したFlowの情報を、後から変更できるPrefectのAPIを使用します。

実行したFlow IDを取得し、update_flow_run メソッドで変更することができます。

https://docs.prefect.io/2.10.18/api-ref/server/api/flow_runs/#prefect.server.api.flow_runs.update_flow_run

以下のサンプルコードでは、Flow処理内部で実行時間を取得し、Flow実行名を後から設定しています。

rename custom flow run name for prefect 2.0

Prefect Cloud上でも指定したFlow実行名に変更されていることを確認できます。

rename flow name on Prefect Cloud

参考