Airflowでよく使う便利なOperator一覧

2023.11.20

はじめに

新田

こんにちは、分析基盤や分析のお仕事をしている新田です。
Airflowは分析基盤でよく使われるワークフロー管理プラットフォーム、僕なりに例えると「リッチなcrontab」ですね。
いろいろなタスクをOperatorというクラスを使って定義するのですが、今回はよく使う便利なOperatorをまとめてみました。

DummyOperator

何も実行しないOperatorです。
まだ具体的なタスクの実装ができていない場合の仮のタスクとしたり、複数タスクをまとめる終了タスクとして使用したり、条件分岐したあとの何もしないタスクが必要な時に使用したりします。
また、DAGを途中から実行したいことがよくあるのですが、その時に選択できる中間ポイントとして使うこともあります。
以下の例だとstart_bのDownstreamをClearするとtask_b1とtask_b2だけ実行することができます。
最初にAirflowを勉強していた時は何の意味があるんだ? と思っていましたが、たくさんのタスクのフローを定義しているとほぼ必ず必要になるOperatorです。

BashOperator

Bashを実行するOperatorです。
システムコマンドや.shファイルを実行することができます。
Bashゆえに何でもできてしまうのですが、Bashばかり使っていると逆にわかりにくくなってしまうことも多いです。

PythonOperator

Python関数を実行するOperatorです。
Pythonで処理を書けるのでいろいろなことができます。
もちろんpandasやnumpy, sklearnを使ってなにかすることができますが、あまりに重いような処理はGKEなど別の環境で動かしたほうがいいこともあります。

BranchPythonOperator

Python関数の実行結果によって、次に実行するべきタスクを自動で決定することができます。
条件付きのワークフローが必要な場面でとても便利です。
python_callableで渡す関数は、次に実行するタスクのtask_idを返却するようにします。
例えば、対象のデータソースが更新されているかチェックして、更新していなければ実行をスキップするようにして、効率の良いデータパイプラインを作ることができます。

BigQueryInsertJobOperator

BigQuery関係のクラスはいろいろあるのですが、古くて非推奨のOperatorもあるので注意が必要です。
このOperatorは、BigQueryのジョブを実行するオペレーターです。
destinationTableを指定すると、そのテーブルにクエリ結果が入ります。
destinationTableを指定せずに、CREATE OR REPLACE TABLE などを使うこともできますし、DELETE文などを実行することもできます、SQLでどのようなクエリでも実行することができます。

TaskGroup

Operatorとは少し異なりますが、タスクをグループにまとめて管理しやすくするクラスです。
Airflow2.0で追加されました。
TaskGroupをクリックすることで、開閉することができますし、マウスをホバーするとTaskGroupの中のタスク全てでの実行時間も表示されます。
特にタスクの数が多い場合や複雑なDAGの場合でもシンプルに見せることができますし、コードもシンプルになりますのでとても便利です。

まとめ

今回は、Airflowでよく使うOperatorを紹介しました。
他にも、SlackWebhookOperator、EmailOperator、GCSToBigQueryOperator、ComputeEngineStartInstanceOperator、EC2StopInstanceOperatorなどとてもたくさんのOperatorがあります。できることが山ほどあって楽しいですね。
適切なOperatorを選んで構築することで、タスクの可読性や保守性が大幅に向上します。
Airflowを使うときは、やりたいことにあったOperatorを選び、効率的なワークフローを構築してみてください。