スケーラブルなログ活用“完全ガイド”

はじめに：ログはなぜ重要か

数十件のジョブから数百件にスケールするのは、いくつかの理由で困難ですが、そのひとつが「オブザーバビリティ（可観測性）」です。オブザーバビリティとは、ログやメトリクス、トレースなどのコンポーネントを分析することでシステムの状態を理解する能力のことです。これは、大規模なシステムに限らず、少数のパイプラインしか持たない小規模なデータチームにとっても同様に重要です。

特にSparkのような分散コンピューティングエンジンは、信頼性の高いモニタリングやデバッグ、エスカレーション手順の整備が難しい傾向があります。

この中でも「ログ」は、最もシンプルで、かつ効果的なオブザーバビリティの手段と言えるでしょう。しかし、ログを1ジョブずつクリックしてスクロールして確認する方法では、スケーラビリティがありません。時間がかかる上に、内容の解釈も難しく、ワークフローに関する専門知識が求められる場合も多いです。

データパイプラインに成熟したログ標準を組み込んでおかないと、エラーやジョブ失敗時のトラブルシューティングに時間がかかり、結果として高額な障害対応、非効率なエスカレーション、アラート疲れなどの問題を引き起こします。

本ブログでは、以下の内容を解説します：

• 単純なprint文から脱却し、適切なロギングフレームワークを構築するステップ

• Sparkのlog4jログをJSON形式で出力するべきタイミング

• ログを解析・検索しやすくするために、クラスターのログを一元管理すべき理由

• Databricksで独自のワークスペースにセットアップできる、カスタマイズ可能なAI/BIダッシュボードの構築方法

重要なアーキテクチャ上の考慮点

技術的ソリューションの内訳

本番レベルのログ監視を実現するには、標準化が鍵となります。理想的には、このソリューションは数百～数千のジョブ／パイプライン／クラスターに対応できる必要があります。

このソリューションの完全な実装は、以下のリポジトリをご参照ください：
https://github.com/databricks-industry-solutions/watchtower

ログの一元配信のためのVolumeの作成

まず、ログ用の中央ファイルストレージとして使用するUnity Catalog Volumeを作成します。
DBFSは推奨しません。なぜなら、DBFSでは同等のデータガバナンスレベルが提供されないためです。

各環境（例：開発、ステージング、本番）のログは、異なるディレクトリやVolumeに分けることを推奨します。これにより、より細かくアクセス制御が可能になります。

Volumeの作成方法は、以下のいずれかです：

• Databricks UIから作成

• Databricks Asset Bundle (AWS | Azure | GCP) 内で作成

• または今回の例では、Terraformを用いて作成します。

ボリュームにREAD VOLUMEとWRITE VOLUMEの権限があることを確認してください（AWS | Azure | GCP）。

クラスターログ配信の設定

ログを置く場所ができたので、クラスターがログをこの目的地に配信するように設定する必要があります。これを行うには、クラスター上で計算ログ配信（AWS | Azure | GCP）を設定します。

再度、UI、Terraform、または他の好みの方法を使用します。今回はDatabricks Asset Bundles (YAML)を使用します：

クラスターまたはジョブを実行すると、数分以内にCatalog Explorerから該当のVolumeにアクセスし、ログファイルが保存されていく様子を確認できます。
クラスターID（例：0614-174319-rbzrs7rq）を名前としたフォルダが作成され、その中に各種ログのグループごとにフォルダが作成されます：

• driver：ドライバーノードからのログ。最も重要なログが含まれます。

• executor：クラスター内の各Sparkエグゼキュータからのログ。

• eventlog：「Event Log」タブで確認できる、クラスターの開始、終了、スケール変更などのイベントログ。

• init_scripts：クラスターで初期化スクリプトが設定されている場合に生成されるフォルダ。クラスター内の各ノードごとにサブフォルダが作成され、各ノードで実行された初期化スクリプトに対応する stdout および stderr のログが保存されます。

標準の徹底：クラスターポリシーの活用

ワークスペース管理者は、可能な限り標準構成を徹底すべきです。
そのためには、ユーザーによる自由なクラスター作成を制限し、あらかじめログ設定を固定値として定義した クラスターポリシー（AWS | Azure | GCP）を付与することが推奨されます。

以下のように、クラスターのログ設定をポリシーで明示的に固定することで、標準化を強制できます：

これらの属性を「固定値（fixed）」として設定することで、適切なVolumeの保存先が自動的に設定され、ユーザーが設定を忘れたり変更したりすることを防げます。

そのため、Asset BundleのYAMLで cluster_log_conf を明示的に設定する代わりに、使用するクラスターポリシーのIDを指定するだけで済むようになります。

print() だけでは不十分

開発中のちょっとしたデバッグには print() 文が便利なこともありますが、本番環境では以下の理由から不向きです：

• 構造がない：print() は非構造のテキストを出力するため、大規模なログの解析やクエリ処理に不向きです。

• 文脈情報が乏しい：タイムスタンプ、ログレベル（INFO、WARNING、ERRORなど）、発生元のモジュール名やジョブIDなど、トラブルシューティングに必要な情報が欠けています。

• パフォーマンスへの影響：print() を多用するとSparkの評価処理を引き起こし、パフォーマンスに悪影響を与える可能性があります。さらに print() は標準出力（stdout）に直接書き込まれるため、バッファリングや最適な処理が行われません。

• 詳細度の制御ができない：print() にはログの詳細度（verbosity）を制御する仕組みがなく、多すぎたり少なすぎたりする原因になります。

その点、Scala/Java（JVM）の Log4j や、Pythonの 標準loggingモジュール などの適切なロギングフレームワークを使えば、上記の課題をすべて解決できます。
これらのフレームワークでは、ログレベルや詳細度の設定、JSONなど機械可読な形式での出力、柔軟な出力先の指定などが可能です。

Sparkにおける stdout / stderr / log4j の違いにも注意

• stdout：ドライバーノードのJVMからの標準出力バッファ。print() 文や通常の出力はここに書き込まれます。

• stderr：標準エラーバッファ。例外やスタックトレースなどがここに出力され、多くのロギングライブラリもデフォルトで stderr を使用します。

• log4j：log4j ロガーを使用して書き込まれたログメッセージ専用で、stderr にも表示されることがあります。

Pythonの場合

Pythonでは、標準の logging モジュールをインポートし、JSONフォーマットを定義して、ログレベルを設定します。

Spark 4（または Databricks Runtime 17.0 以降）では、PySpark に組み込まれた構造化ロガー が利用可能になっています：
https://spark.apache.org/docs/latest/api/python/development/logger.html

以下の例は、ロガーインスタンスを pyspark.logger.PySparkLogger に置き換えることで、PySpark 4 に適応できます。

このコードの大部分は、PythonのログメッセージをJSON形式に整形するためのものです。
JSONは半構造化データであり、人間にも機械にも読みやすいため、後ほどログを取り込んでクエリを実行する際にその利便性を実感できるでしょう。
もしこの整形処理を省略した場合、メッセージのどの部分がログレベルで、どれがタイムスタンプ、どれが実際のメッセージなのかを推測するために、複雑で非効率な正規表現に頼ることになるかもしれません。

もちろん、このコードをすべてのノートブックやPythonパッケージに毎回書くのは冗長です。
このような**定型コード（ボイラープレート）**は、ユーティリティコードとしてまとめ、以下のような方法でジョブに読み込ませることで、重複を避けることができます：

• Pythonモジュールとしてワークスペース上に保存し、Workspace File Import（AWS | Azure | GCP）を使って、各ノートブックの冒頭で実行する

• Pythonのwheelファイルとしてパッケージ化し、クラスターにライブラリとして読み込む（AWS | Azure | GCP）

Scalaの場合

Scalaでも同じ考え方が適用されますが、こちらでは Log4j（より正確にはその抽象レイヤーである SLF4j）を使用します。

UIでドライバーログを表示すると、Log4jの下にINFOとWARNのログメッセージを見つけることができます。これは、デフォルトのログレベルがINFOであるため、DEBUGとTRACEのメッセージは書き込まれません。

ただし、Log4j のログは JSON 形式にはなっていません！
この点については、次のセクションで対処方法をご紹介します。

Spark Structured Streamingのためのロギング

ストリーミングジョブにおいて、ストリーミングソースやシンクのメトリクス、クエリの進捗状況などの有用な情報を取得するには、Sparkの StreamingQueryListener を実装する方法があります。

次に、作成したクエリリスナーをSparkセッションに登録します：

SparkのStructured Streamingクエリを実行すると、Log4jのログに次のような出力が表示されるようになります。
（※この例では、Deltaをソースおよびシンクとして使用しています。出力される詳細なメトリクスは、ソースやシンクの種類によって異なります）

Spark の Log4j ログ設定

以下の初期化スクリプトを使用して、これをJsonTemplateLayoutに変更します：

この初期化スクリプト（init script）は、PatternLayout を JsonTemplateLayout に置き換えるだけの内容です。
なお、init script はクラスター内のすべてのノード（ドライバーノードだけでなくワーカーノードも含む）で実行されます。
ただしこの例では、ログの詳細度（冗長性）の観点からドライバーログのみを設定対象としており、また後続のログ取り込み処理でもドライバーログのみを使用する前提です。

なお、ワーカーノードでも同様の設定ファイルが以下のパスに存在します：
/home/ubuntu/databricks/spark/dbconf/log4j/executor/log4j.properties

必要に応じてこのスクリプトに処理を追加することも可能ですし、cat $LOG4J2_PATH で元の設定ファイルの内容を確認すれば、変更点の把握や編集がしやすくなります。

次に、この init script を Unity Catalog の Volume にアップロードします。
整理しやすくするために、前に作成したログ用の Volume を使い回すのではなく、別の Volume を新たに作成します。
Terraform を使えば、以下のようにして作成できます：

この設定により、Volumeが作成され、初期化スクリプト（init script）が自動的にアップロードされます。

しかし、クラスターにこの init script を使用するよう設定する必要がまだ残っています。
先ほどはクラスターポリシーを使ってログの保存先（Log Delivery Destination）を固定しましたが、同じようにこの init script についてもポリシーで強制することで、Sparkログを常に構造化されたJSON形式で出力させることができます。

そのために、先ほどのクラスターポリシーのJSONに以下の設定を追加します：

ここでも「固定値（fixed）」として設定することで、init script が常にクラスターに適用されることが保証されます。

これで、先ほどのSparkコードを再実行してみると、Log4jセクションのドライバーログがすべてきれいなJSON形式で出力されていることが確認できます！

ログの取り込み

ここまでで、単なる print() 文から脱却し、構造化されたロギングに切り替え、Sparkのログと統合し、それらを中央のVolumeにルーティングするところまで実現できました。
この時点でもすでに、Catalog Explorer や Databricks CLI を使ってログファイルを閲覧・ダウンロードすることができ、便利です：

databricks fs cp dbfs:/Volumes/watchtower/default/cluster_logs/cluster-logs/$CLUSTER_ID . --recursive

しかし、このロギングハブの真の価値が発揮されるのは、ログを Unity Catalog のテーブルに取り込むときです。
これによりログ管理のループが完結し、クエリの実行や集計、さらにはパフォーマンス問題の検出まで、あらゆる分析が可能になります。
これらの詳細については、これから順に説明していきます！

ログの取り込みは Lakeflow Declarative Pipelines を使えば非常に簡単で、Auto Loader とメダリオンアーキテクチャを活用して増分読み込みを実現します。

Bronzeログ

最初のテーブルは、ドライバーログの生データを読み込むためのブロンズテーブルです。
このテーブルでは、ログ本体に加えて、ファイル名、サイズ、パス、最終更新時刻といった追加のカラムも付加されます。

Lakeflow Declarative Pipeline の Expectations（期待値）機能（AWS | Azure | GCP）を活用することで、ネイティブなデータ品質モニタリングも自動的に得られます。
この後に登場する他のテーブルでも、同様のデータ品質チェックが行われる様子をご紹介します。

Silverログ

次に作成する Silverテーブル は、より重要な役割を果たします。
このテーブルでは、ログの各行のテキストをパースし、ログレベル、タイムスタンプ、クラスターID、ログの出力元（stdout / stderr / log4j） などの情報を抽出します。

※補足：できる限りJSON形式でのロギングを設定していても、起動時に実行される他のツールなどからのログ出力には、非構造なプレーンテキストが含まれる可能性があります。
こうした多くのログは stdout に出力されており、Silverテーブルへの変換処理では、まずJSONとしてメッセージのパースを試み、失敗した場合のみ正規表現での解析にフォールバックすることで、柔軟なパース処理を実現しています。

コンピュートID

パイプラインの最後のテーブルは、Databricksのシステムテーブルを基に構築されたマテリアライズドビューです。
このテーブルには、各ジョブ実行時に使用された コンピュートID（クラスターID） を保存し、特定のログを出力したジョブIDを後から簡単に特定できるようにする目的があります。

なお、1つのジョブが複数のクラスターを使う場合や、ジョブクラスターではなくSQLウェアハウス上でタスクが実行されるケースもあるため、あらかじめこのリファレンステーブルを作成しておくことには大きな意義があります。

パイプラインのデプロイ

パイプラインは、UI、Terraform、またはアセットバンドル内からデプロイすることができます。アセットバンドルを使用し、以下のリソースYAMLを提供します：

AI/BI ダッシュボードでログを分析する

最後に、ジョブ・ジョブ実行・クラスター・ワークスペースを横断してログデータをクエリできるようになります。
Unity Catalog 管理テーブルの最適化により、これらのクエリは高速かつスケーラブルに実行できます。

以下にいくつかの例を見てみましょう。

エラー件数の多い順（Top N Errors）

このクエリでは、頻出するエラーを特定することができます。これにより、優先的に対応すべきエラーの傾向を把握し、エラーハンドリングの改善や対応手順書（Runbook）の整備に活用することができます。

エラーによるトップNジョブ

このクエリは、観察されたエラーの数でジョブをランク付けし、最も問題のあるジョブを見つけるのに役立ちます。

AI/BIダッシュボード

これらのクエリを Databricks の AI/BI ダッシュボード に組み込むことで、
すべてのログを検索・フィルタリングし、よくある問題の検出やトラブルシューティングを行うための中央インターフェースとして活用できるようになります。

この例のAI/BIダッシュボードは、このソリューションのすべてのコードとともにGitHubで利用可能です。

実際のユースケース

リファレンスダッシュボードで示したように、このようなロギングソリューションが対応できる実践的なユースケースは数多くあります。たとえば：

• 単一のジョブについて、すべての実行ログを横断的に検索する

• すべてのジョブを対象にログを検索する

• 最も頻出しているエラーの分析

• エラー件数が多いジョブの特定

• パフォーマンス上の問題や警告の監視
  　- GC Overhead（ガベージコレクションのオーバーヘッド）
  　- Sparkのスピル処理（ディスクへの退避）

• ログ内の個人情報（PII）の検出

現実の現場では、担当者がジョブの実行ごとに手動で切り替えながらエラーを確認し、どのアラートを優先すべきか判断できないというケースが多くあります。
しかし、堅牢なログ基盤と標準的なテーブル構造を整備することで、頻出エラーを簡単にクエリで特定し、優先順位をつけることが可能になります。

たとえば、1件のジョブがOutOfMemoryエラーで失敗しており、別の10件のジョブが、サービスプリンシパルから誤ってSELECT権限が剥奪されたことで失敗しているとします。
アラートが大量に届いている中でも、こうした仕組みがあれば、担当者は「優先すべきは権限エラーのほうだ」と即座に判断でき、まずその問題解決に取りかかることで、10件のジョブを迅速に復旧させられます。

また、同じジョブの複数の実行を比較するためにログを確認するケースもよくあります。
たとえば、「各バッチ実行で 'batch completed' というログが出力された時刻」と、「そのバッチが呼び出したAPIのリクエストスループットのグラフ」とを突き合わせるようなケースです。
ログを取り込んでおけば、テーブルに対してジョブIDや任意のジョブ実行IDを指定してフィルタし、一つひとつの実行に手動で入らずとも簡単に比較分析ができます。

運用上の考慮点

• クラスターのログは 5分おきに配信され、1時間ごとにgzip圧縮されて保存先に格納されます。

• Unity Catalog 管理テーブルに Predictive Optimization と Liquid Clustering を併用することで、ログテーブルのパフォーマンスを最大限に引き出せます。

• 生のログファイルは無期限に保存する必要はありません（これがクラスターログ配信のデフォルト動作です）。Declarative Pipelines では cloudFiles.cleanSource オプションを使って、指定した保持期間後にファイルを削除できます。保持期間は cloudFiles.cleanSource.retentionDuration で定義します。また、クラウドストレージ側のライフサイクルルールを使っても対応可能です。

• エグゼキュータログの取り込み設定も可能ですが、ほとんどのエラーはドライバーノードに伝播されるため、通常は不要です。

• 取り込んだログテーブルを使って、Databricks SQL Alerts（AWS | Azure | GCP）を設定することで自動アラートの仕組みを追加することも検討してください。

• Lakeflow Declarative Pipelines には独自のイベントログも存在し、パイプラインの実行状況を監視・確認する用途で使用できます。このイベントログもUnity Catalogに書き出すことが可能です。

外部ツールとの連携と実現できること

お客様によっては、Loki、Logstash、AWS CloudWatch などの一般的なロギングツールとログを連携させたいというニーズもあるでしょう。
それぞれに認証や設定、接続要件はありますが、基本的にはクラスターの初期化スクリプト（init script）を使って設定を行い、ログ転送用のエージェントを実行するという点では、共通のパターンに従います。

重要なポイントまとめ

最後に、本ガイドの主要なポイントを振り返ります：

• 本番環境では print 文ではなく、標準化されたロギングフレームワークを使用すること

• クラスター単位の初期化スクリプト（init script）で Log4j の設定をカスタマイズすること

• クラスターのログ配信設定を行い、ログを一元管理すること

• Predictive Optimization と Liquid Clustering を併用した Unity Catalog 管理テーブルを使用し、最高のパフォーマンスを実現すること

• Databricksでは、ログの取り込みと強化（エンリッチ）を行い、より高度なログ分析が可能になること

次のステップ

本ガイドで紹介したログ活用を、ぜひ本番環境に取り入れてみてください。
完全なソリューションの詳細は、以下の GitHub リポジトリをご覧ください：
👉 https://github.com/databricks-industry-solutions/watchtower

DatabricksのDelivery Solutions Architects (DSAs)は、組織全体のデータとAIのイニシアチブを加速します。彼らはアーキテクチャリーダーシップを提供し、プラットフォームをコストとパフォーマンスの観点から最適化し、開発者体験を向上させ、プロジェクトの成功を推進します。DSAsは、初期デプロイメントと本番環境のソリューションの間のギャップを埋め、データエンジニアリング、技術リーダー、エグゼクティブ、その他のステークホルダーを含むさまざまなチームと密接に連携して、カスタム化されたソリューションとより早い価値提供を確実にします。DSAからカスタム実行プラン、戦略的ガイダンス、データとAIの旅を通じてのサポートを受けるためには、Databricksアカウントチームにお問い合わせください。