Apach Airflowなどワークフローエンジンおけるプラクティスとアンチパターン
私は Data Engineer的な業務を長くやっている傍ら、パイプラインのプラテクティスについて考えることがよくあります。
代表的なものはAirflow, digdag, Luigiなどです。
（また私の経験上話がETLや機械学習向けの処理に少し偏るかもしれませんがご了承ください。）
しかし、意外とこのあたり記事やナレッジが少ない領域だと思います。
それにも関わらずアンチパターンも少なくありません。
例えば datetime.now()で実行時間を取る書き方です。依存関係手前の処理が膨らんだ時、日付や時間を跨ぐ危険性もありますし、再実行時どうするんだよ？という気持ちです。
今回は自身の経験をまとめようと思います。
また、唯一見つけた web上の記事である Airflowの Best Practiceも参考にして書こうと思います。
そして本題に入る前に、アンチパターン詰め込みワークフローの一例も書いてみたので、こちらをご覧になりながらぜひ読んでみてください。
再実行可能性・冪等性
時間の取得方法に気を付ける
datetime.now的なものを書かない。
datetime.now()には二つの危険性があります。
- 再実行時に時間指定できないこと。
- もうひとつはそれに至るまでの処理が膨らんだ時に、想定している時間・日付を不本意に跨いでしまう可能性があります。
日付時間に関する定義はjob実行側で行い、コードは引数でそれを受け取るようにしましょう。
(Database書き込み時のcreateAt的なものは例外です。）
データへの書き込みは何度やっても同じように
例えばデータをテーブルに書き込む際、insert文だけ書いているケースは危険です。
再実行時に多くのワークフローエンジンはコードレベルでの途中から実行はサポートしていません。
insertを繰り返し重複行の発生や一意エラーを引き起こす可能性があります。
insert対象のデータは必ず一歩手前で削除する、upsert(merge insert)等で処理することを徹底しましょう。
通知（特にエラー通知)
書きやすさであったり、重要度であったり、思いつきなのか、よくわからない特定の場所にだけエラー通知を書いているケースはよくありません。
必ず全体に対して仕込みましょう。
システム管理者向けにはまず、job全体を監視するように設計する
よくわからない一つのタスク、またはコード内の一つの処理に対してエラー通知を自己実装するのはナンセンスです。エラーはjob全体のどこでおきるかわかりません。
まずは
複雑なロジックの分離
反復処理(for loop)や分岐(if)はあまりワークフローの中で使うべきではないと考えています。
ワークフローは一つ一つのタスクに対して時間制限や単体での通知、利用する計算エンジンの設計などを持たせられることが多く、複雑な処理を実現するには too muchな設定を引きずることになります。
あくまでも命令系統であることを意識して、複雑な処理はコード側に分離しましょう。
命令、変数の一方通行（有向非巡回グラフ）
これは個人的な意見かもしれませんが、jobの中にあるタスクからjob側に変数や設定を戻すことはあまり良いと考えていません。
理由はいくつかあります。
- job全体に作用する変数にも関わらず、どこで定義されたか探すこと自体がナンセンス
- 並列処理されるタスクの中で同時に定義されるとどうなるのか不確定な要素が増える
- 処理の途中に書き換えられるとdebugの際に読みに行きにくい上、直列のみのフローならまだしも、依存関係が見にくい
つまりは、命令及び変数定義は有向非巡回グラフ/Directed acyclic graph的であるべきだということです。ここでの方向は二つの観点があります。 - ワークフローエンジンとタスク（例えばAirflowとspark)
- タスクの順序、つまりは単純に実行させる順番
これらは親子関係を保持すべきです。特に一つ目は重要で、ワークフローエンジンはあくまでも指令・命令系統であり、その中の変数がタスク側から書き換えられる場合には最新の注意が必要だということです。
また、例外は少なくないと思います。ワークフロー全体にまたがる複雑な変数をワークフロー側の標準関数では実装できないケースなどです。
疎結合
別にワークフローエンジンに限った話ではなくシステム全般における話だとは思いますが、ワークフローも疎結合に設計した方が良いです。
特にdependencieを操作する際には、一定小さなjobに分けて、jobをcallするようにするといいと思います。（サポートされていれば）
Secretや環境変数の管理場所を統一する
Airflowの中にはいくつかの変数やSecretが管理できるにも関わらず、無作為にSecret Managerから取ったり、分散しているケースがよくありません。
- 管理自体が大変になる
- 環境が変わった際に誤ったソースから読み取るようなコーディングリスクがある
環境依存する変数を受け渡す場合注意が必要です。
いまだに悩むこと
- 複雑な時間処理はどうすべき？例えば日本の祝日などはjob側の標準的な関数で実装が難しいです。自分は最低限できる限りの処理をjobで行った上で、(例えば平日のみ実行）、該当しない日付においてはjobを早々にcloseするように処理しています。ただし実行に一々コストもかかりますし、logも無駄にNレコード増えるので、もっとイカしたやり方はないかと悩んでいます。
まとめ
実は全て自分で思いつきで書いて、答え合わせとしてAirflow Best Practiceを読んだのですが、結構内容が一致していてびっくりしました。
今回はAirflowのベストプラクティス以外あまり資料は見つけられなかったです。
ナレッジの少ない領域だと思うので皆さんにご意見も聞けたら嬉しく思います。
いいねいただけるとはげみになります。
お読みいただきありがとうございました。
Discussion