【ITニュース解説】A Green Spark Job Isn't Enough: Building a Fail-Closed Pipeline Health Monitor in Python

現代のシステム開発において、データパイプラインはビジネスの意思決定を支える重要な基盤だ。特に、大量のデータを段階的に処理し、分析に役立つ形式に変換するSparkのような分散処理フレームワークを使ったパイプラインは一般的だが、単に「ジョブが成功した」という表面的な結果だけでは、システムが本当に健全であるとは言えない場合がある。例えば、パイプラインの中間処理で一時的にエラーが発生して回復したとしても、その過程でデータの整合性が損なわれたり、処理が遅延したりする可能性がある。また、システムが一部の処理を開始したものの、完了イベントを送信せず、見かけ上は動いているように見えても実際には停止している「ストール状態」に陥ることも考えられる。さらに、ヘルスチェックのレポートが不完全な状態でダッシュボードに表示され、誤解を招く「緑色の成功表示」となってしまう危険性もある。

小売業界のメディア分析パイプラインでは、キャンペーン指標データの遅延がパフォーマンス低下と誤解されたり、最終段階のデータ更新が不完全なのに成功と表示されたりする事態は、ビジネス判断を狂わせる。このような課題に対処するため、この記事では、Pythonを使って、より信頼性の高い「フェイルクローズド」なパイプラインヘルスモニターを構築する方法が解説されている。

「フェイルクローズド」とは、システムが不確かな状態にある場合、安全側に倒して「異常」と判断する設計思想だ。つまり、何らかの異常や不明な点があれば、安易に「正常」と判断せず、明確な「正常である」という証拠がない限りは「異常」とみなすことで、誤った安心感を排除する狙いがある。このモニターの最も重要な設計原則の一つは、パイプラインの実際のデータ処理（実行）と、その処理が健全であるかどうかの判断（ヘルス解釈）を完全に分離することだ。パイプラインの各処理ステップは、オーケストレーターと呼ばれる制御役によって順次実行され、各ステップの結果として構造化されたイベントを記録する。このイベントは、例えば、最初のデータ変換（BronzeからSilver）と次のデータ変換（SilverからGold）という段階で構成されるパイプラインにおいて、どのステージがいつ開始し、いつ終了し、成功したか失敗したか、といった情報を時系列で記録する。

このモニターは、Sparkセッションを起動したり、処理の再試行を決定したりすることなく、記録されたイベントを読み込んでパイプラインの現在の健全性を再構築する。この分離によって、モニターは非常に軽量に動作し、ダッシュボードは複雑なログを解析する必要なく、モニターのシンプルな出力を直接利用できるという実用的な利点がある。

このシステムでは、単なるキーワード検索ではなく、明確な「イベントコントラクト」に基づいて健全性を判断する。イベントコントラクトとは、イベントデータの形式と内容に関する取り決めを指す。具体的には、パイプラインのオーケストレーターは、pipeline_events.jsonlというファイルに、1行1JSON形式でイベント情報を追記していく。このイベントデータは、run_id（パイプライン実行の識別子）、stage（処理段階）、attempt（試行回数）、status（状態、例：started、succeeded、failed）、timestamp_utc（協定世界時でのタイムスタンプ）などの必須情報を含む。失敗イベントの場合はreturncode（終了コード）やerror（エラーメッセージ）も含まれる。

JSON Lines形式は、イベントが順次追記され、モニターが各行を個別に処理できるため、この用途に適している。ここで重要なのは、「アテンプト（試行）」と「ラン（実行）」という概念の違いだ。例えば、ステージ1の初回アテンプトが失敗しても、システムが自動的に再試行し、2回目のアテンプトで成功する可能性がある。この場合、1つのアテンプトは失敗したが、ステージ1全体としては最終的に成功したと見なせる。しかし、パイプラインが複数のステージで構成されている場合、たとえステージ1が成功しても、ステージ2が失敗すればパイプライン全体の「ラン」は成功とは言えない。モニターはこれらの関係性を正確に理解し、最終的な健全性を判断する。

モニターは、イベントストリームをそのまま信用するのではなく、宣言的に「健全である」と判断する前に厳密な検証を行う。summarize_events()という機能は、存在しないまたは無効なrun_id、ステージ番号、アテンプト回数、ステータス、タイムゾーン情報を持たないタイムスタンプ、時系列順が乱れているイベント、重複したアテンプト開始、アテンプト番号のスキップ、先行する失敗がないのに再試行が行われた記録、開始イベントがないのに完了イベントがあるケース、成功なのにゼロ以外の終了コードを持つイベント、ステージ1が成功する前にステージ2が開始された場合など、多岐にわたる不正なイベントデータを積極的に拒否し、エラーとして報告する。

これらのルールに違反する、つまり「不正なログ」は、黙って無視されたり、空の実行として扱われたりすることなく、直ちにエラーとして処理される。これは「フェイルクローズド」ポリシーの具体例であり、ダッシュボードに緑色の「正常」表示を出すためには、システムが正常であるという積極的かつ検証済みの証拠が求められる。パイプライン全体の「ラン」レベルの健全性は、期待される各ステージの最新の状態に基づいて計算される。全ての期待されるステージが成功して初めて、そのランは「成功」と見なされる。そうでなければ、失敗したステージがあれば「失敗」、処理が停滞している場合は「ストール」、現在実行中のステージがあれば「実行中」、全てが完了していない場合は「未完了」というように、優先順位に従って状態が決定される。

モニターは、単なる「失敗」だけでなく、「ストール」状態も正確に検出する。ストールとは、プロセスが開始イベントを発行したものの、その後、完了イベントを含むいかなるイベントも生成せずに停止してしまった状態を指す。モニターは、未完了のアテンプトが設定された閾値（例: 600秒）を超えてイベントを生成しない場合に、「ストール」として分類する。この閾値はstale_after_secondsとして設定可能だ。ただし、ストールは完了イベントの「不在」から推測されるものであり、必ずしもプロセスの「死」が証明されたわけではない点には注意が必要だ。非常に長い時間かかる正当なジョブには、より長い閾値設定や、定期的に生存を示すハートビートプロトコルが必要になる場合がある。

モニターは、パイプラインの健全性に関するスナップショットをJSON形式で出力する。この出力には、スキーマバージョン、生成日時、最新のランID、実行回数、状態ごとのカウント、そして各ランとステージの詳細な記録が含まれる。このJSONファイルは、小さな運用ダッシュボードやビジネスインテリジェンスツールに取り込んで、パイプラインの現在の状態を可視化するために利用できる。

このJSONファイルを書き出す際も、非常に堅牢な方法が採用されている。直接既存のレポートファイルを上書きするのではなく、まず一時ファイルに内容を書き込み、内容が完全に書き込まれてディスクに同期されたことを確認してから、os.replace()という機能を使って元のファイルを新しい一時ファイルにアトミック（不可分）に置き換える。この方法は、書き込み中にシステムがクラッシュした場合でも、元の健全なレポートが破損したり、不完全なデータで上書きされたりするのを防ぐことができる。イベントの検証に失敗した場合には、そもそも新しいファイルが生成されないため、以前の健全な状態のファイルが保持され、誤解を招く部分的なレポートが出力されることはない。

このモニターは、自動化されたワークフローで非常に役立つように設計されている。コマンドラインから実行でき、--fail-on-unhealthyというオプションを使用すると、コマンドの終了コード自体がパイプラインの健全性を示す。例えば、全てのランが成功していれば終了コードは0、一つでも失敗、ストール、実行中、未完了のランがあれば1、イベントが無効だったりI/Oエラーが発生したりすれば2となる。これにより、CI/CD（継続的インテグレーション/継続的デリバリー）パイプラインのステップとして組み込んだり、定期的なチェックとして自動実行したりする際に、スクリプトがこの終了コードを基に自動的に後続の処理を進めるか停止するかを判断できる。

このモニターの設計では、成功パスだけでなく、様々な状態遷移（例えば、再試行後に成功するケース、恒久的に失敗するステージ、実行中からストールへの移行、無効なイベントの順序など）を網羅するテストが非常に重視されている。合成データを用いたテスト環境で、これらの複雑なシナリオが正しく処理されることを検証することで、システムの信頼性が高められている。

しかし、この設計が解決しない領域も明確にされている。このモニターは「オペレーショナルヘルス」、つまりパイプラインが技術的に正しく動作しているかを監視するものであり、「データ品質」や「データの正確性」を保証するものではない。例えば、パイプラインが全てのステージを問題なく完了しても、出力されるキャンペーン指標自体が間違っている可能性は残る。そのため、別途データ品質チェックやKPI（重要業績評価指標）の妥当性検証が必要になる。また、分散システムにおけるロック、イベントブローカー、複数ワーカーの連携、ハートビートによる生存監視、アラートのルーティング、ログの保持ポリシーなどもこのモニターの範囲外だ。本番環境でこのシステムを利用する場合には、ジョブの所有権の明確化、ハートビートの導入、データ保持期間の設定、イベント取り込みの冪等性（何度実行しても結果が同じであること）、構造化されたアラート通知、そして遅延したり順序が乱れたりしたイベントに対する明確な処理ポリシーを追加で検討する必要がある。

結論として、信頼できる「緑色の正常ステータス」とは、単に最後のコマンドが「完了」と出力したからといって推測されるものではなく、順序立てられ、厳密に検証された証拠に基づいて「証明」されるものだ。このPythonで構築された、軽量で徹底的にテストされたモニターは、再試行や部分的な実行、そしてストールしたステージといった、これまでは見過ごされがちだった問題状態を可視化する。これにより、すべてのヘルスチェックのために新たなSparkジョブを起動するといったオーバーヘッドなしに、パイプラインの運用健全性を確実に把握できるようになるのだ。