Prefect2.0のレンダリング機能に感動

はじめに

こんにちは！D2Cのデータサイエンティスト（MLエンジニア）の須田です。
最近Prefectを構築し、色々触ってみたところレンダリング機能に感動しました。
この気持ちをみなさんにも共有したく記事を書きます！

Prefectのレンダリング機能とは

flowを実行した際にArtifactsとして記録される情報をリンク/マークダウン/テーブルの形式でUI上に表示できる機能です。文章だけだとイメージしづらいと思うので今回はコードと画像で紹介します。

早速Prefectのレンダリング機能を使ってみる（3種）

※Prefectが適切に使用できる環境があることを前提として進めます。
※Kubernetes(helm)を使用して環境構築をした過去記事はこちら

今回は簡単に試せるように政府CIOポータルというサイトのオープンデータのcsv（オープンデータに取組済の都道府県・市区町村の一覧資料csv）を使用して、試します。

1. データをマークダウン形式にし、Artifactsとして表示する

from prefect import flow, task
from prefect.artifacts import create_markdown_artifact

@task
def markdown_task():
    na_revenue = 500000
    markdown_report = f"""# 都道府県統計情報レポート

このレポートでは、日本の都道府県に関する統計情報を提供します。データは2022年1月12日時点のものです。

## 人口と面積統計

以下は各都道府県の2020年度推計人口と面積（平方キロメートル）に関する統計情報です。

| 都道府県      | 人口（2020年度推計） | 面積（平方キロメートル） |
|--------------|----------------------|--------------------------|
| 北海道       | 5381733              | 83424.65                 |
| 青森県       | 1308265              | 9645.67                  |
| 岩手県       | 1279594              | 15275.01                 |
| 宮城県       | 2333899              | 7286.31                  |
| 秋田県       | 995842               | 11637.56                 |
| ...          | ...                  | ...                      |
| 熊本県       | 1774345              | 7405.08                  |
| 大分県       | 1166338              | 6341.52                  |
| 宮崎県       | 1104069              | 7735.31                  |
| 鹿児島県     | 1616617              | 9189.62                  |
| 沖縄県       | 1444501              | 2276.66                  |

## 結論

このレポートでは、各都道府県の2020年度推計人口と面積に関する統計情報を提供しました。データセットからの追加情報や具体的な分析が必要であれば、さらに詳細なレポートを作成できます。統計情報を活用して地域や都道府県の比較分析を行うなど、さまざまな用途に応じてデータを活用できます。
"""
    create_markdown_artifact(
        key="cio-opendata-report", #keyには小文字のアルファベット、数字、およびダッシュ（ハイフン）しか含むことができないので注意
        markdown=markdown_report,
        description="政府CIOポータルから取得したデータでレポートを作りました",
    )

@flow()
def markdown_example():
    markdown_task()

実行

markdown_example()

結果

05:24:50.690 | INFO    | prefect.engine - Created flow run 'cordial-fulmar' for flow 'markdown-example'
05:24:50.698 | INFO    | Flow run 'cordial-fulmar' - View at http://prefect-server.prefect.svc.cluster.local:4200/flow-runs/flow-run/9910221b-7de4-4b84-af3f-2e46be69d222
05:24:50.798 | INFO    | Flow run 'cordial-fulmar' - Created task run 'markdown_task-0' for task 'markdown_task'
05:24:50.801 | INFO    | Flow run 'cordial-fulmar' - Executing 'markdown_task-0' immediately...
05:24:52.231 | INFO    | Task run 'markdown_task-0' - Finished in state Completed()
05:24:52.276 | INFO    | Flow run 'cordial-fulmar' - Finished in state Completed('All states completed.')
[Completed(message=None, type=COMPLETED, result=UnpersistedResult(type='unpersisted', artifact_type='result', artifact_description='Unpersisted result of type `NoneType`'))]

UIを確認する

Dashboard>Flows>Runsから実行したflowを選択します。

該当のflowのtaskを選択します。

Artifactsのタブをクリックします。（Artifactsが表示されない場合は何回かリロードしてください）

以下のようにマークダウンの枠が表示されるのでクリックします。

きちんとレポートされていますね！

今回はオープンデータの中身の情報をマークダウンとして表示しましたが、機械学習のモデルを作成するタスクがあった場合はモデルの精度や使用した特徴量やパラメーターなどをレポートとして表示したら良いのではと思いました。

create_markdown_artifact()に限らず以下で紹介する関数もkeyには小文字のアルファベット、数字、およびダッシュ（ハイフン）しか含むことができないので注意してください。

2. リンクを作成し、Artifactsとして表示する

from prefect import flow, task
from prefect.artifacts import create_link_artifact

@task
def link_task(): #リンクをそのまま表示する
        create_link_artifact(
            key="government-cio-opendata", 
            link="https://cio.go.jp/sites/default/files/uploads/documents/digital/opendata_lg_pref_list_20220112.csv",
            description="## 政府CIOポータルからデータを取得しました",
        )

@task
def textlink_task(): #テキストリンクとして表示する
        create_link_artifact(
            key="government-cio-opendata-2",
            link="https://cio.go.jp/sites/default/files/uploads/documents/digital/opendata_lg_pref_list_20220112.csv",
            link_text="政府CIOポータルからデータを取得しました",
        )

@flow
def link_example():
    link_task()
    textlink_task()

実行

link_example()

結果

05:47:31.252 | INFO    | prefect.engine - Created flow run 'ingenious-petrel' for flow 'link-example'
05:47:31.255 | INFO    | Flow run 'ingenious-petrel' - View at http://prefect-server.prefect.svc.cluster.local:4200/flow-runs/flow-run/21d03200-3bd3-4a89-af48-804722ca21cd
05:47:31.362 | INFO    | Flow run 'ingenious-petrel' - Created task run 'link_task-0' for task 'link_task'
05:47:31.366 | INFO    | Flow run 'ingenious-petrel' - Executing 'link_task-0' immediately...
05:47:31.616 | INFO    | Task run 'link_task-0' - Finished in state Completed()
05:47:31.656 | INFO    | Flow run 'ingenious-petrel' - Created task run 'textlink_task-0' for task 'textlink_task'
05:47:31.658 | INFO    | Flow run 'ingenious-petrel' - Executing 'textlink_task-0' immediately...
05:47:31.844 | INFO    | Task run 'textlink_task-0' - Finished in state Completed()
05:47:31.897 | INFO    | Flow run 'ingenious-petrel' - Finished in state Completed('All states completed.')
[Completed(message=None, type=COMPLETED, result=UnpersistedResult(type='unpersisted', artifact_type='result', artifact_description='Unpersisted result of type `NoneType`')),
 Completed(message=None, type=COMPLETED, result=UnpersistedResult(type='unpersisted', artifact_type='result', artifact_description='Unpersisted result of type `NoneType`'))]

UIを確認する

実行したflowからArtifactsを確認するとリンクが以下のように表示されます。
今回はリンクをそのまま表示するものとテキストリンクとして表示するの2つのArtifactsが生成されていますね。

リンクをそのまま表示した場合

テキストリンクとして表示した場合

このリンクを押すとcsvをローカルに保存できたりします。データを出力してローカルで中身を確認したい、同じファイルや出力をチームで共有したい場合は良さそうです。

3. データをテーブルにし、Artifactsとして表示する

次は政府CIOの登録済み都道府県一覧資料のcsvを取得し、テーブルとしてUIに表示するcreate_table_artifact()を使用してみます。

from prefect import flow, task
import pandas as pd
from prefect.artifacts import create_table_artifact


@task
def get_csv_as_df(path):
    df = pd.read_csv(path, encoding='shift-jis')
    return df

@task
def create_artifact(data, data_name):
    data_dict = data.to_dict(orient='records')

    return create_table_artifact(
        key=f'{data_name}',
        table=data_dict,
        description=f'{data_name}の中身をテーブル化',
    )

@flow
def table_example():
    csv_path= "https://cio.go.jp/sites/default/files/uploads/documents/digital/opendata_lg_pref_list_20220112.csv"
    df = get_csv_as_df(csv_path)
    create_artifact(df,"cio-dict") 

実行

table_example()

結果

05:50:43.471 | INFO    | prefect.engine - Created flow run 'tactful-hog' for flow 'table-example'
05:50:43.473 | INFO    | Flow run 'tactful-hog' - View at http://prefect-server.prefect.svc.cluster.local:4200/flow-runs/flow-run/685af735-b242-4632-832d-4fd7cc58992e
05:50:43.554 | INFO    | Flow run 'tactful-hog' - Created task run 'get_csv_as_df-0' for task 'get_csv_as_df'
05:50:43.557 | INFO    | Flow run 'tactful-hog' - Executing 'get_csv_as_df-0' immediately...
05:50:43.879 | INFO    | Task run 'get_csv_as_df-0' - Finished in state Completed()
05:50:43.916 | INFO    | Flow run 'tactful-hog' - Created task run 'create_artifact-0' for task 'create_artifact'
05:50:43.918 | INFO    | Flow run 'tactful-hog' - Executing 'create_artifact-0' immediately...
05:50:44.117 | INFO    | Task run 'create_artifact-0' - Finished in state Completed()
05:50:44.179 | INFO    | Flow run 'tactful-hog' - Finished in state Completed('All states completed.')
[Completed(message=None, type=COMPLETED, result=UnpersistedResult(type='unpersisted', artifact_type='result', artifact_description='Unpersisted result of type `DataFrame`')),
 Completed(message=None, type=COMPLETED, result=UnpersistedResult(type='unpersisted', artifact_type='result', artifact_description='Unpersisted result of type `UUID`'))]

UIを確認する

flowから該当のArtifactsをクリックすると以下のようにテーブルが表示されます。

csvで取得したデータの中身をUIからテーブルとして確認することができましたね。
どんなレコード、カラムが入っているか目で見て確認したい時は使ってみると便利だと思います！

さいごに

どうでしたか？私が所属している部署では主にAirflowを使用しているのですが、Airflowにはレンダリング機能がないため、Prefectのこの機能にワクワクしました。今回はオープンデータで試しましたが実務でも有効活用していきたいです。

ここまでお読みいただきありがとうございました！

参考URL

https://docs.prefect.io/latest/api-ref/prefect/artifacts/
https://cio.go.jp/policy-opendata/index.html#catalogsite