概要

ComposerのAirflowのカスタムTriggerを作成した。

色々と難しかったけど、使えるとリソース消費を抑えられて、コスト抑制できるのでおススメ。

なぜカスタムTriggerを使ったのか

定期的にAPIでBigQueryテーブルの更新状況をチェックして更新されていたら、テーブルに依存する別のテーブルを更新したかった。

AirflowのWorkerでその処理を実行しようとすると、テーブルが更新されるまでCPUとメモリを占有してしまい、お金がかかる。

更新状況チェック対象のテーブルが大量にあるのと、数時間というレベルで待機が必要になる。

そういったユースケースに対して、導入されたのがTriggerプロセス。

AirflowのOperatorでdeferメソッドを実行すると、Triggererプロセスに処理が渡され、非同期で複数の処理をまとめて実行してくれる。

Triggerの処理が完了すると、呼び出し元のOperatorに処理が戻されてタスクが完了する。

Triggererプロセスに処理が渡されている間は、タスクはdefferedステータスになり、Workerで使用されていたリソースは解放されるので、テーブルが更新されるまで待機するといったタスクでもリソース消費を抑えて処理できる。

今回はGCP ComposerのAirflowでカスタムTriggerを動かした。

DAGとは違って、ComposerバケットにDAGのPythonファイルを入れても動かない。

TriggererプロセスはAirflowの起動時に読み込まれて実行が開始される。

そのため、PyPIパッケージとしてComposerにインストールする必要がある。

ComposerにカスタムPyPIパッケージをインストールする方法は色々あるが、今回はArtifact Registryの標準リポジトリに生成したPythonパッケージをアップロードして、そこからComposerにインストールした。

その話はまた別途投稿したい。

基本的には下記を参考にすれば、問題なく実装できると思う。

何か不明な点や解決できない問題があれば、メッセージください。