DagsterでYearlyPartitionsDefinitionを実装する

Dagster

概要

Dagsterでbackfillの単位として、パーティションという概念があります。
docs.dagster.io
その中でも時間単位で制御したい場合、下記が提供されています。

  • HourlyPartitionsDefinition
  • DailyPartitionsDefinition
  • WeeklyPartitionsDefinition
  • MonthlyPartitionsDefinition
  • TimeWindowPartitionsDefinition

時間、日、週、月までの時間単位はDagster側で提供されていますが、年次パーティションは現在実装されていませんでした。
年次パーティションはどのように実装したらよいか調べていたら同じ問題を抱えていた人のISSUEを見つけました。
github.com
年次単位のように長い期間のパーティション単位や2週間ごとのように中途半端な期間のパーティションなど、Dagsterが提供しているパーティション単位以上にカスタムしたい場合は、TimeWindowPartitionsDefinitionを使い、自身で設定する必要があるようです。

実装

下記の実装で1970年から1年ごとのパーティションが実装できました。

@asset(
    partitions_def=TimeWindowPartitionsDefinition(
        start=datetime(1970, 1, 1),  # 1970年1月1日を起点
        fmt="%Y",   # 表示は1970のように年単位で
        cron_schedule="0 0 1 1 *"  # 毎年1月1日0時0分ごとに
    )
)

上記assetを含むJobのbackfillは次のようになりました。
2024年は現在進行中なので2023年までしか出ていません。

実行している年のパーティション値がほしい場合は、次のように取得可能です。

partition_next_year: int = context.partition_time_window.start.year + 1