ContentsAirflowPluginsとはどんな時にAirflowPluginsを使うのか新規Plugin作成方法Plugin確認方法どんなComponentをPluginに追加することができるのか1.hooksを使用して独自のHookを作成する。2.macrosを使用してカスタムmacroを作成する。3.flask_blueprintsを使用してAirflowのFlaskにflask.BluePrint製のページを作成する。4.appbuilder_viewsを使用してAirflowのFlaskにflask_appbuilder.BaseView製のページを作成する。UIのメニューにセクションとリンクを作成する。6.global_operator_extra_linksを使用して全ての種類のOperatorにLinkを作成する7.operator_extra_linksを使用して任意の種類のOperatorにLinkを作成する。8.timetablesを使用してCRONでは表現できないスケジュールを作成する終わり参考AirflowPluginsとは#Airflowをカスタマイズするために追加できるPlugin。どんな時にAirflowPluginsを使うのか#例えば以下のようなときにAirflowPluginsを使う。AirflowのメタデータのDBのコンテンツを表示するViewを作成したいとき。指定された時間内において任意数のDAGが失敗時にAirflowの機能を監視しカスタムアラートを送信するアプリケーションを作成したいとき。外部データツールのファイルやログに動的にリンクする詳細ビューをUIに表示したいとき。新規Plugin作成方法#「plugins_folder」へPythonファイルを用意し、Pythonファイル内でairflow.plugins_manager.AirflowPluginを拡張するクラスを定義すれば新規Pluginを作成できる。定義したAirflowPluginクラスには以下を実装する。そのほかPluginに追加したいComponentについて列挙していく。上記場合、Plugin名が「empty」であるPluginを作成している。「plugins_folder」の確認の仕方ちなみに「plugins_folder」はどこで確認できるのかというのだが、「コマンドを利用する」か「cfgファイルを見る」の2通りで確認できる。基本的にデフォルトでは$AIRFLOW_HOME/pluginsになっているはず。xxxxPlugin確認方法#現在追加されているPluginは何か、Pluginに定義されてあるComponentは何かを確認する方法だが、「UIから確認する方法」と「CLIを利用して確認する方法」がある。WebServerを再起動しなくてもすぐに反映されるのは後者の方法。「UIから確認する方法」UIのAdmin->Pluginsを使用する。pluginsを実行する。__init__.plugin_macroどんなComponentをPluginに追加することができるのか#例えば以下のようなComponentが現在ある。種類としてUIへの機能追加からScheduling、DAG作成する上での機能追加などなどがある。詳細についてはドキュメントを確認すると良い。CRONでは表現できないスケジュールを作成するそのほか「executorを使用してCustomExecutorを作成する」「listenerを使用してイベント検知を作成する」などできそう。今回は上記のComponentのみ順に試してみます。1.hooksを使用して独自のHookを作成する。#AirflowのHookとは#外部システムとのインタフェースの役割をしてくれるようなもの。以下のように書いて一貫したメソッドを提供することを意図としてある。PostgresHook.get_hook(conn_id)pg_hook.copy_exper()詳細:https://airflow.apache.org/docs/apache-airflow/stable/concepts/connections.htmlplugins_folder配下のファイル#今回は次のようなファイル構造で新規独自のHookを作成していきます。['PluginHook']hooks/airflow_hook_template.pyairflow.hooks.base.BaseHookを拡張して書く。pass上記の場合、特に何もしないHook。#webserverの再起動が必要。dagのファイルからimportしてみる#plugins_folderは環境変数PYTHONPATHとして自動的にbuildされている(https://airflow.apache.org/docs/apache-airflow/stable/modules_management.html#built-in-pythonpath-entries-in-airflow)ので,直接plugins_folder配下のディレクトリを指定する形でimportできる。UIからDAGを確認するなどをしてimportErrorが出ていなければ正常に動作している。PluginHook2.macrosを使用してカスタムmacroを作成する。#macroとは#JinjaTemplateを使用してTask内で利用できる変数のようなもの。{{execution_date}}')詳細:https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.htmlplugins_folder配下のファイル#plugins_folder配下に__init__.pyを作成してそこに記載する形で書いていく。__init__.py関数に作成するmacroについて定義していく。1した形で返すということをしている。#webserverの再起動が必要。dagのファイルから利用してみる#dagのファイル作成したmacroはmacros.<plugin_name>.<macro_name>で取得できる。test_plugins()log上記DAGを実行してみたLogだが、引数1に対しoutputが2となっているのでmacroが効いていることを確認できる。23.flask_blueprintsを使用してAirflowのFlaskにflask.BluePrint製のページを作成する。#flask.BluePrintとは#Flask内でアプリケーションの機能を複数のファイルに分割する方法。ソースコードが長すぎる場合に、機能ごとにファイルを分割し管理しやすくするために利用する。plugins_folder配下のファイル#plugins_folder配下に__init__.pyを作成してそこに記載する形で書いていく。__init__.pyairflow.plugins_manager.AirflowPluginを拡張してかく。IP>:<port>/直下のSubpathとなる。[bp]上記の場合、Subpathが/bp_testであるページを追加しており、ページを開くとweatherAPIを使用して任意の市の今日の日付と天気を取得し表示するようなことをしている。#webserverの再起動が必要。作成したページにアクセスしてみる。IP>:<port>/bp_testへアクセスすると以下のようなページが表示される。4.appbuilder_viewsを使用してAirflowのFlaskにflask_appbuilder.BaseView製のページを作成する。#flask_appbuilder.BaseViewとは#FlaskのViewのうち、よりきめ細かくセキュリティ(メソッド,アクセス)を設定できるView。plugins_folder配下のファイル#plugins_folder配下に__init__.pyを作成してそこに記載する形で書いていく。__init__.pyAirflowのメニューバーのセクションにViewのLinkを設定する形で書いていく。test_page’と表示されることを期待している。#webserverの再起動が必要。作成したページにアクセスしてみる。#期待通りのメニューバーとセクションができている。Viewへアクセスしてみるとページが正常に確認できる。UIのメニューにセクションとリンクを作成する。#plugins_folder配下のファイル#plugins_folder配下に__init__.pyを作成してそこに記載する形で書いていく。notesにViewへのLinkを設定しており、Linkをアクセスするとhref先のページへ遷移されることを期待している。#webserverの再起動が必要。作成したページにアクセスしてみる。#期待通りのメニューバーとセクションができている。セクション先をアクセスするとhttps://467tn.comへ遷移される。6.global_operator_extra_linksを使用して全ての種類のOperatorにLinkを作成する#AirflowのOperatorLinkとは#Taskの詳細画面につけるボタンLink(うまい表現がわかりません)。基底がBaseOperatorLinkとなっており、ボタンLinkをClickすると当クラスのget_link:functionが実行されるわけだが、get_link:functionには引数として当Taskのoperator(BaseOperator)とti_key(airflow.models.taskinstance.TaskInstanceKey)が渡される。このOperatorLinkを使えば、Operatorが外部インタフェースと通信するようなOperatorだった場合に外部インタフェース先にあるLogなどをti_keyをもとに表示するようなことができます。標準でprovideされているOperatorLinkのうち、BigQueryConsoleLinkは、BigQueryOperatorを使用してBigQueryに投げたLogをGoogleCloudConsoleから当job_idを指定して確認することができるような仕様をしているOperatorLinkです。詳細:https://airflow.apache.org/docs/apache-airflow/stable/howto/define_extra_link.htmlplugins_folder配下のファイル#plugins_folder配下に__init__.pyを作成してそこに記載する形で書いていく。__init__.pyBaseOperatorLinkを拡張して書く。name:propertyに渡した値がボタンLink名get_link:functionに定義した処理はボタンLinkをClickした時に動作する挙動。戻り値としてstrでLinkPathを記載。docsというボタンLinkを作成しており、ボタンLinkをClickするとhttps://airflow.apache.org/へ遷移する。#webserverの再起動が必要。作成したLinkを確認してみる。#任意のTaskの詳細を開くとOperatorの種類に限らず期待通りのボタンLinkが作られていることを確認できる。ボタンLinkをクリックするとhttps://airflow.apache.org/へ遷移する。7.operator_extra_linksを使用して任意の種類のOperatorにLinkを作成する。#OperatorLinkとは#「6.global_operator_extra_links」での説明と同様なので割愛。plugins_folder配下のファイル#plugins_folder配下に__init__.pyを作成してそこに記載する形で書いていく。__init__.pyBaseOperatorLinkを拡張して書く。name:propertyとget_link:functionの仕様については「6.global_operator_extra_links」での説明と同様。operators:List[BaseOperator]にLinkを反映したいOperatorを列挙する。docsというボタンLinkを作成しており、ボタンLinkをClickするとhttps://kubernetes.io/へ遷移する。このLinkはKubernetesPodOperatorを使用したTaskにしか反映されないことを意図としている。#webserverの再起動が必要。作成したLinkを確認してみる。#KubernetesPodOperatorのあるDAGを開いて確認してみる。KubernetesPodOperatorのTaskに期待通りのボタンLinkが作られていることを確認できる。ボタンLinkをクリックするとhttps://kubernetes.io/へ遷移する。PythonOperatorのTaskにはボタンLinkが作られていない。8.timetablesを使用してCRONでは表現できないスケジュールを作成する#個人的な本題だったので少し他よりも丁寧に書く。Timetableとは#AirflowのDAG内での「Cronで表すことができないようなスケジュール」を決めるためのTimeTable。DAGに渡すtimetable引数で指定できる。「Cronで表すことができないようなスケジュール」とは例えば以下のようなスケジュール。10:00のみ実行するなど毎日異なる時間に実行する水曜日は10:00、木曜日は03:00に実行するなどグレゴリオ暦に従わないスケジュール標準でprovideされているTimeTableEventsTimetableCronTriggerTimetable詳細:https://airflow.apache.org/docs/apache-airflow/stable/concepts/timetable.html独自のTimetable作成方法について#Airflow2.2から独自のTimeTableを作成することができるようになった。airflow.timetables.base.Timetableを拡張すれば良いのだが、以下の関数を含める必要がある。手動実行時のデータ間隔で返す独自のTimetableに指定する関数を作成する。#今回は以下のような仕様のTimeTableを作ってみます。1日に2回,06:00(1)と16:30(2)にジョブを実行する。実行時に取得するデータの間隔は以下。data_interval:DataInterval)で返すparameterに入ってくるrestriction:TimeRestrictionとはDAGをスケジュールできる期間に関する制限(earliest:DateTime,latest:DateTime,catchup:bool)を持つクラス。初回実行ではない場合(a)前回実行のデータ間隔のStartが06:00の場合(b)今回の実行は1に該当するnext_startは前回の当日の16:30next_endは前回の翌日の06:00前回実行のデータ間隔のStartが16:30の場合(c)今回の実行は2に該当するnext_startは前回の翌日の06:00next_endは前回の翌日の16:30初回実行の場合(d)DAGをスケジュールできる期間のうちの最早い日を実行日(next_start)として(catchup=Falseの場合、現在の日付を実行日(next_start)とする)next_startは実行日の06:00next_endは実行日の16:30またこの時DAGの終了日(restriction.latest)がある場合、その日を越したnext_endを実行しないことを期待している。end:Datetime)で返す。end=end)補足すると現在時刻が06:00~16:30の間(a)1の実行とされる。startは前日の16:30、endは当日の06:00現在の時刻が16:30~23:00の間(b)2の実行とされる。startは当日の06:00、endは当日の16:30それ以外、深夜の00:00~05:59の間(c)2の実行とされる。上記で書いた関数をそのまま移植していきます。#webserverとschedulerの再起動が必要。(TimeTableはschedulerも読み込む必要があるためschedulerも再起動します。再起動しないとTimetableNotRegisteredErrorが出ます。)作成したTimeTableを試してみる。適当なDAGのtimetable引数に作成したMyTimeTable:TimeTableを渡す。次の実行は01/05.16:30次の次の実行は01/06.06:00。その時のデータ間隔は01/05.16:30~01/06.06:00終わり#以上AirflowのPluginについて見ていきました。TimeTableについては、AirflowのScheduleについて再度理解する機会になってよかったです。時間があるときにlistenerやexecutorのカスタムも試してみたいです。