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
メソッドで変更することができます。
以下のサンプルコードでは、Flow処理内部で実行時間を取得し、Flow実行名を後から設定しています。
rename custom flow run name for prefect 2.0
Prefect Cloud上でも指定したFlow実行名に変更されていることを確認できます。