Executorの話こんにちは。Teamのhatoneがお届けします。Teamは、データサイエンティスト・マーケティング・会計チームetcの多岐にわたる社内データのパイプラインの構築・運用を行っています。Executorを用いたデータパイプラインの例をお話しながらご紹介していきたいと思います。このようなプロセスが繰り返されていきます。膨大なデータ量を如何に解釈して扱っていくかを模したフレームワークとして、DIKWピラミッドというモデルがあります。hierarchyと呼ばれています。DIKWピラミッドの図このDIKWピラミッドにおける最下層に位置するDataは、未整理状態の生データやログのことを示しています。この例として、各オフィスの当日の最高気温をざっくりまとめた表があるとします。2019年12月05日この状態では、下記のように各項目の内容が未整理状態となっています。Cityカラムが漢字/英語表記が混在Temperatureカラムで気温の表記が摂氏(°C)と華氏(°F)が混在Dateのカラムがタイムゾーン不明瞭でフォーマットがばらついているこのような上の表の各要素が未整理のまま収集されている状態がDataです。こちらを、ちょっと整理してみましょう。混在していた各カラムの内容を、一定の基準で整理していきます。DIKWピラミッドにおけるInformationは、明確に定められた基準の上で整理・処理された状態を示します。DataとInformationを同じような意味で用いられることがあるのですが、より整形されたDataがInformationと呼ばれているため、ピラミットの1つ上の層にInformationが位置しています。Altoの方が高かったが、12月5日の時点では東京の方が高かった」といった事実を見ることができるようになります。このInformationから導き出される傾向や規則性を導出されたものが、DIKWピラミッドにおけるKnowledgeになります。頂点であるWisdomは、導き出されたKnowledgeに基づいて人により下される判断のことそのものを示します。Engineeringの仕事は、このDataを過不足無く蓄えること、DataからInformationへの変換・蓄積する作業がメインと言えます。Pipelineを構築運用しています。Executorを使ったパイプイライン構築例そんな日々のDataをInformationへ変換して蓄積していく例として、「8TB程度あるdumpログ(非正規化されている文字列込み)をBigQueryへ取り込む」や「カラム結合が駆使されて見目麗しく整っているExcelスプレッドシートの中腹部分からいい感じにレコードを取り出して、BigQueryへ格納する」など様々あります。Composer上でいくつか稼働させてデータパイプラインを運用してます。社内で使われるLookerというBIツールが連携しているため、最終的に蓄積する場所がBigQueryになっています。そのような背景から、”Airflow上でPythonスクリプトを使ってデータを処理し、整形済みデータをGCSへ格納、BigQueryへロードしていく”という手をよく使っています。Executorを用いたパイプラインの例をご紹介していきます。1.10.0から導入された比較的新しいExecutorで、タスクインスタント毎に新しいポッドを作成して実行してくれるものです。下記が、特定のレポートをAPI経由で取得して再整形してBigQueryへロードするパイプラインの一例です。アップロードしています。Registoryへ登録しておきます。Pythonスクリプトはユースケースに応じて変わってしまうのですが、Shellスクリプトはだいたい下記のようなパターンになります。gs://GCS_DESTINATION/${TARGET_DATE}/これをAirflow上で呼び出すためには、下記のようにDAGのOperaterを書きます。is_delete_operator_pod=True,)image_pull_policyオプションを’Always’にしておくことで、Pod起動時に常にImageをPullしてくるようになります。開発検証時のみis_delete_operator_podオプションをFalseにしておくことで、Podが何らかのエラーで正しい挙動をしなかった場合に削除されずデバッグがしやすくなります。この2つを設定しておくことで、開発効率が少しだけ良くなるおすすめのオプションです。Bでは、GCSに上がった整形済みデータを、BigQueryへとロードしていきます。GoogleCloudStorageToBigQueryOperatorを下記のように指定しています。ようにAirflowのMacroを記述しています。Cは、他の処理が何らかのエラーが発生した場合にSlackチャンネルへアラートをあげるためのものです。BigQueryOperatorStepDは、StepBの月毎を全て結合し、任意のカラムのみにしたものを出力する例です。trigger_rule=TriggerRule.ALL_DONE,)このような流れで、特定のレポートをAPI経由で取得して再整形してBigQueryへロードするDataをInformation化する作業を実現しています。Operatorを使う手がありました。この場合は独自のライブラリを利用したい場合にはAirflow本体の環境へライブラリを追加する必要が発生していました。あるデータパイプイラインにて特定のバージョンのライブラリを使いたいが、別データパイプラインでは同ライブラリの別バージョンを利用したいようなケースの場合ではバッティングしてしまうという問題がありました。Imageベースで実行するため他環境のAirflowへの移行も容易で、有用な点が非常に多いと感じています。さいごに時折、想像の範囲を軽く超えたデータパイプライン開発のチケットが切られることがあります。Engineeringは、ビッグデータならぬビックリデータを如何に実装していくか?というパズル問題のような思考から、単純に大きなサイズのデータを扱うこと、Airflowクラスタのメンテナンスなど、データをいい感じに扱うための技術スキル総合格闘技感が楽しいところです。今後も、処理できるデータの幅と芸風を広げていくために、精進してまいります。