【Streamlit】高度なロギングテクニックと実践ガイド

こんにちは、JS2IIUです。
今回は、PythonのWebアプリケーションフレームワークであるStreamlitにおけるロギングの活用方法について解説します。
ロギングは、エラーの検知やアプリの動作確認、パフォーマンスチューニングなど、多くの場面で不可欠な技術です。しかし単純にログを取るだけでなく、ログの一貫性やパフォーマンスへの配慮、さらにはCI環境での運用まで考慮すると、かなり奥が深い領域でもあります。
この記事では、Streamlitアプリでの基本のロギング方法から、カスタムロガーの作成、非同期ロギング、ログの構造化、分析・可視化、複数モジュールでのログ統合、さらには自動テストや継続的インテグレーション（CI）における実践的なロギング運用まで、実用的なコード例とともにステップバイステップでご紹介します。

1. はじめに：ロギングの重要性と役割

ロギングは、アプリケーションの状態やエラー情報を記録し、問題発生時の原因解析や動作確認に役立ちます。特にWebアプリケーションでは、ユーザーの操作ログや処理結果を記録し続けることが、安定的なサービス運用に不可欠です。

Streamlitは手軽にインタラクティブアプリが作れますが、その分、内部の状態確認やエラー解析が難しくなりがちです。きちんとログを取ることで、開発・運用の双方で問題をスピーディに特定できるようになります。

本記事では以下の内容を順に解説します。

• Streamlitでの基本ロギング
• カスタムロガーの設定方法
• 非同期ロギングの導入
• ログメッセージの構造化
• パフォーマンスへの影響評価
• ログ分析と可視化手法
• 複数モジュール間のログ統合
• 効率的なトラブルシューティング
• 自動テスト・CI環境でのロギング活用

2. ロギングの基本

Pythonの標準ライブラリloggingを使い、Streamlitアプリ内にログを出力する基本的方法をご紹介します。

ステップ1：importとlogger作成

import streamlit as st
import logging

# loggerの作成とレベル設定
logger = logging.getLogger('streamlit_app')
logger.setLevel(logging.DEBUG)

# handlerの設定（コンソール出力）
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)

# フォーマッターの設定
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
console_handler.setFormatter(formatter)

# loggerにhandlerを追加
if not logger.hasHandlers():
    logger.addHandler(console_handler)

コードの説明

1. import streamlit as st
   • Streamlitライブラリを st という名前でインポートします。Streamlitアプリを作成する際には必須です。
2. import logging
   • Pythonの標準ライブラリである logging をインポートします。プログラムの動作を記録するための機能を提供します。
3. logger = logging.getLogger('streamlit_app')
   • streamlit_app という名前のロガーを作成（または取得）します。アプリケーション内の異なるモジュールで同じ名前を指定すれば、同じロガーを共有できます。
4. logger.setLevel(logging.DEBUG)
   • このロガーが処理するログの最低レベルを DEBUG に設定します。DEBUG は最も詳細なレベルで、これ以降 INFO, WARNING, ERROR, CRITICAL の順に重要度が高くなります。これにより、DEBUG レベル以上のすべてのログが処理対象となります。
5. console_handler = logging.StreamHandler()
   • ログの出力先としてコンソール（ターミナル）を指定するための「ハンドラ」を作成します。
6. console_handler.setLevel(logging.DEBUG)
   • このコンソールハンドラが出力するログの最低レベルを DEBUG に設定します。
7. formatter = logging.Formatter(...)
   • ログの出力フォーマット（書式）を定義します。
     • %(asctime)s: ログが出力された日時
     • %(name)s: ロガーの名前（この場合は streamlit_app）
     • %(levelname)s: ログのレベル（DEBUG, INFOなど）
     • %(message)s: 実際のログメッセージ
8. console_handler.setFormatter(formatter)
   • 先ほど作成したコンソールハンドラに、定義したフォーマットを適用します。
9. if not logger.hasHandlers(): logger.addHandler(console_handler)
   • このロガーにまだハンドラが設定されていない場合（if not logger.hasHandlers()）に限り、作成したコンソールハンドラを追加します。これにより、同じハンドラが重複して追加されるのを防ぎます。

ステップ2：Streamlitアプリ内でのログ出力

def main():
    logger.info('アプリが起動しました')
    st.title("Streamlit Logging Example")

    user_input = st.text_input("文字を入力してください")
    if user_input:
        logger.debug(f'ユーザー入力: {user_input}')
        st.write(f'あなたの入力: {user_input}')

if __name__ == "__main__":
    main()

ポイント

• logger.info(), logger.debug(), logger.error() などのレベルに応じたログ出力ができる
• StreamHandlerは標準出力にログを表示するため、ストリームリットの端末ログ上に確認可能

3. カスタムロガーの作成と設定

より柔軟にログを管理したい場合は、自分でカスタムロガーを作成し、ファイル出力や複数ハンドラーの設定を行います。

ステップ1：ロガー設定関数を作成

import os

def setup_custom_logger(name, log_file='app.log', level=logging.DEBUG):
    logger = logging.getLogger(name)
    logger.setLevel(level)

    # 既にhandlerがある場合は重複防止
    if logger.hasHandlers():
        logger.handlers.clear()

    # コンソール用ハンドラー
    ch = logging.StreamHandler()
    ch.setLevel(level)
    ch_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    ch.setFormatter(ch_formatter)

    # ファイル用ハンドラー
    fh = logging.FileHandler(log_file, encoding='utf-8')
    fh.setLevel(level)
    fh_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    fh.setFormatter(fh_formatter)

    logger.addHandler(ch)
    logger.addHandler(fh)

    return logger

コードの解説

1. 関数定義

def setup_custom_logger(name, log_file='app.log', level=logging.DEBUG):

• setup_custom_logger という名前の関数を定義しています。
• 引数（ひきすう）:
  • name: ロガーの名前。必須です。
  • log_file='app.log': ログを保存するファイル名。指定がなければ app.log が使われます。
  • level=logging.DEBUG: 記録するログの最低レベル。指定がなければ DEBUG（最も詳細なレベル）になります。

1. ロガーの取得とレベル設定

    logger = logging.getLogger(name)
    logger.setLevel(level)

• 引数 name で指定された名前のロガーを取得します。
• このロガーが処理する最低レベルを、引数 level で設定します。

1. ハンドラの重複防止

    # 既にhandlerがある場合は重複防止
    if logger.hasHandlers():
        logger.handlers.clear()

• この関数が複数回呼ばれた場合などに、同じ設定（ハンドラ）が何度も追加されてログが重複して出力されるのを防ぎます。
• if logger.hasHandlers(): で、ロガーに既にハンドラが設定されているかを確認します。
• もし設定されていれば logger.handlers.clear() で全ての設定を一度リセットします。これは非常に堅牢な設計です。

1. コンソール用ハンドラの設定

    # コンソール用ハンドラー
    ch = logging.StreamHandler()
    ch.setLevel(level)
    ch_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    ch.setFormatter(ch_formatter)

• logging.StreamHandler(): ログをコンソール（ターミナルやコマンドプロンプト）に出力するためのハンドラを作成します。
• ch.setLevel(level): ハンドラが出力するレベルを設定します。
• logging.Formatter(...): コンソールに出力する際のフォーマットを定義します。こちらは比較的シンプルな形式になっています。
• ch.setFormatter(...): ハンドラにフォーマットを適用します。

1. ファイル用ハンドラの設定

    # ファイル用ハンドラー
    fh = logging.FileHandler(log_file, encoding='utf-8')
    fh.setLevel(level)
    fh_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    fh.setFormatter(fh_formatter)

• logging.FileHandler(log_file, encoding='utf-8'): ログをファイルに書き出すためのハンドラを作成します。
  • log_file で指定されたファイルに書き込みます。
  • encoding='utf-8' は、日本語などのマルチバイト文字が文字化けしないようにするための重要な設定です。
• fh.setLevel(level): ハンドラが出力するレベルを設定します。
• logging.Formatter(...): ファイルに書き出す際のフォーマットを定義します。こちらはロガー名 (%(name)s) を含む、より詳細な形式になっています。
• fh.setFormatter(...): ハンドラにフォーマットを適用します。

1. ロガーにハンドラを追加して返す

    logger.addHandler(ch)
    logger.addHandler(fh)

    return logger

• logger.addHandler(): 作成したコンソール用 (ch) とファイル用 (fh) の両方のハンドラをロガーに追加します。これにより、1回のログ出力命令で、コンソールとファイルの両方に記録されます。
• return logger: 全ての設定が完了したロガーオブジェクトを返します。

ステップ2：ロガーを使う

logger = setup_custom_logger('streamlit_custom_logger', 'streamlit_app.log')

def main():
    logger.info('アプリ開始')
    st.title("カスタムロガーの例")

    val = st.number_input("数値を入力")
    if val:
        logger.debug(f'入力された数値: {val}')
    st.write(f"あなたの入力値は {val} です。")

if __name__ == "__main__":
    main()

ポイント

• ログファイルstreamlit_app.logにログが追記される
• 複数のハンドラーでコンソールとファイル出力を両立可能
• logger.handlers.clear()で複数実行時の重複を防止できる

4. 非同期ロギングの導入とその利点

ロギング処理はI/O操作が含まれるため、特に高頻度のログでは処理速度に影響が出ることがあります。非同期ロギングは、メイン処理をブロックせずにログ処理を別スレッドに任せる方法で、パフォーマンス向上が期待できます。

ステップ1：QueueHandler, QueueListenerを使って非同期化

import queue
from logging.handlers import QueueHandler, QueueListener
import threading

def setup_async_logger(name, log_file='async_app.log', level=logging.DEBUG):
    log_queue = queue.Queue(-1)  # 無制限サイズのキュー
    logger = logging.getLogger(name)
    logger.setLevel(level)

    # 基本のファイルハンドラー（Listenerが使用）
    file_handler = logging.FileHandler(log_file, encoding='utf-8')
    formatter = logg

ing.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    file_handler.setFormatter(formatter)

    # QueueHandlerをloggerにセット（非同期でキューに積むのみ）
    queue_handler = QueueHandler(log_queue)
    logger.handlers.clear()
    logger.addHandler(queue_handler)

    # QueueListenerを作成し専用スレッドでハンドリング
    listener = QueueListener(log_queue, file_handler)
    listener.start()

    return logger, listener

コードの解説

1. 必要なモジュールのインポート

import queue
from logging.handlers import QueueHandler, QueueListener
import threading

• queue: スレッド間で安全にデータをやり取りするためのキュー構造を提供します。
• QueueHandler, QueueListener: 非同期ロギングを実現するための専用クラスです。
• threading: QueueListener は内部でスレッドを利用するため、概念の理解に役立ちます。

1. 関数定義とキューの作成

def setup_async_logger(name, log_file='async_app.log', level=logging.DEBUG):
    log_queue = queue.Queue(-1)  # 無制限サイズのキュー

• queue.Queue(-1): ログレコードを一時的に溜めておくためのキューを作成します。-1 はキューのサイズを無制限にすることを意味します。

1. ロガー本体とファイルハンドラの設定

    logger = logging.getLogger(name)
    logger.setLevel(level)

    # 基本のファイルハンドラー（Listenerが使用）
    file_handler = logging.FileHandler(log_file, encoding='utf-8')
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    file_handler.setFormatter(formatter)

• ここは前回までのコードとほぼ同じです。ただし、この file_handler は、この後作成する QueueListener（バックグラウンド処理役）が使うものであり、メインのロガーには直接接続されません。

1. QueueHandlerをロガーに接続

    # QueueHandlerをloggerにセット（非同期でキューに積むのみ）
    queue_handler = QueueHandler(log_queue)
    logger.handlers.clear()
    logger.addHandler(queue_handler)

• QueueHandler(log_queue): これがアプリケーション側の「伝票を置く」役目をするハンドラです。logger.info() などが呼ばれると、このハンドラはログレコードを受け取り、log_queue に追加するだけです。この処理は非常に高速です。
• logger.addHandler(queue_handler): メインのロガーには、この QueueHandler だけを接続します。

1. QueueListenerの作成と起動

    # QueueListenerを作成し専用スレッドでハンドリング
    listener = QueueListener(log_queue, file_handler)
    listener.start()

• QueueListener(log_queue, file_handler): これがバックグラウンドで動く「料理人」です。
  • 第1引数 log_queue: 監視対象のキューを指定します。
  • 第2引数以降 file_handler: キューから取り出したログレコードの処理先（出力先）となるハンドラを指定します。
• listener.start(): リスナーを起動します。これにより、キューを監視するための新しいスレッドが開始され、ログの書き込み処理がバックグラウンドで始まります。

1. ロガーとリスナーを返す

    return logger, listener

• logger: アプリケーション内でログを出力するために使います。
• listener: アプリケーション終了時に、キューに残っているログをすべて書き出し、安全にリスナーを停止させるために必要です。

ステップ2：Streamlitアプリ内への組み込み

logger, listener = setup_async_logger('async_streamlit')

def main():
    logger.info('非同期ロギング開始')
    st.title("非同期ロギングのサンプル")

    text = st.text_input("入力してください")
    if text:
        logger.debug(f'入力内容: {text}')
        st.write(f'はじめまして、{text}さん。')

if __name__ == "__main__":
    try:
        main()
    finally:
        listener.stop()  # アプリ終了時にListener停止

非同期ロギングの利点

• メインスレッドはログの書き込み待ちをしないためUIの応答性が向上
• 大量ログや複数ユーザーでの並列処理に有効
• デッドロック回避にも効果的

5. ログメッセージの構造化と一貫性の確保

ログの解析を効率化するためには、ログメッセージを構造化（例：JSON形式）し、一貫したフォーマットで出力することが重要です。

ステップ1：JSONフォーマッターの導入

pip install python-json-logger

ステップ2：JSONFormatterを使う例

from pythonjsonlogger import jsonlogger

def setup_json_logger(name, log_file='json_app.log', level=logging.DEBUG):
    logger = logging.getLogger(name)
    logger.setLevel(level)

    handler = logging.FileHandler(log_file, encoding='utf-8')
    formatter = jsonlogger.JsonFormatter('%(asctime)s %(name)s %(levelname)s %(message)s %(filename)s %(lineno)d')
    handler.setFormatter(formatter)

    logger.handlers.clear()
    logger.addHandler(handler)
    return logger

json_logger = setup_json_logger('json_logger')

json_logger.info('ユーザーログ', extra={'user_id': 1234, 'action': 'login'})

ポイント

• ログがJSON形式で出力されるため、ログ収集ツール（ELKやFluentdなど）での解析が容易
• extra引数で任意のフィールドを付加できる
• 一貫性が保たれることで監視やトラブル解析が効率化

6. パフォーマンスへの影響評価

ログは便利ですが、多量のログ出力はアプリケーションの性能に影響を与えることがあります。ここでは、簡単な遅延計測でログの影響を評価する方法を紹介します。

測定例（基本同期ロギング）

import time

start = time.perf_counter()

for i in range(1000):
    logger.debug(f'ログメッセージ{i}')

end = time.perf_counter()
st.write(f'同期ロギングにかかった時間: {end - start:.4f}秒')

非同期ロギングでの比較

上記のコードを非同期ロギング環境で試し、処理時間の違いを比較してください。
非同期ロギングは大抵の場合、処理時間を大幅に短縮できます。

7. ログデータの分析と可視化

収集したログは放置せず、適切に分析して改善に活かしましょう。ここでは代表的な分析手法とツールを紹介します。

• ログ解析ツール
  • ELKスタック（Elasticsearch, Logstash, Kibana）
  • Fluentd
  • Grafana Loki
• 分析ポイント例
  • エラーレートの推移
  • ユーザー操作の傾向
  • 応答時間の分布

8. 複数モジュール間でのログ統合

大規模アプリでは複数のモジュール・パッケージからログを出力します。これらのログを1つにまとめ、一貫した形で管理することが重要です。

ポイント

• ルートロガーを作成し、各モジュールは名前空間付きのロガーを取得して出力する
• ハンドラーやフォーマッターはルートロガーで一括管理することで統一感が生まれる

例：ルートロガー設定

# ルートロガー設定（main.pyなど）
root_logger = setup_custom_logger('my_project')

# 各モジュール
import logging

logger = logging.getLogger('my_project.moduleA')

def process():
    logger.info("モジュールAの処理が開始されました")

この方法で、すべてのログはmy_project.logに統合され、管理や分析が容易になります。

9. 効率的なトラブルシューティングとログ活用法

トラブル時にはログから原因特定を迅速に行いたいものです。具体的な活用法をいくつかご紹介します。

• ログレベルで判別：エラーはERRORレベル、詳細な調査はDEBUGを活用
• タイムスタンプで因果関係を追う
• 相関IDを付与し、複数処理を関連付ける

事例：セッションID付与によるトレース

import uuid

def log_with_session_id(logger, message, session_id=None, level=logging.INFO):
    if not session_id:
        session_id = str(uuid.uuid4())
    logger.log(level, f"[session_id={session_id}] {message}")
    return session_id

session_id = log_with_session_id(logger, "処理開始")
# 処理中...
log_with_session_id(logger, "処理中のイベント発生", session_id, logging.DEBUG)

セッションIDが付与されたログは後から絞り込みやすく、問題の流れを追跡しやすくなります。

10. 自動テスト環境でのロギング設定

テスト実行時はログ出力が冗長になりがちですが、適切に設定すればテスト結果の診断に役立ちます。

ポイント

• テスト用ロガーは専用に作成
• ログレベルをWARNING以上に設定し、必要最低限のログに抑える
• テスト失敗時のみ詳細ログを出力する仕組みも便利

pytestでの例

import logging

def setup_test_logger():
    logger = logging.getLogger('streamlit_test')
    logger.setLevel(logging.WARNING)
    if not logger.hasHandlers():
        handler = logging.StreamHandler()
        formatter = logging.Formatter('%(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        logger.addHandler(handler)
    return logger

def test_sample():
    logger = setup_test_logger()
    logger.info("このログは表示されません")
    logger.error("テスト失敗時のエラーログ")
    assert 1 == 1

11. 継続的インテグレーション（CI）でのロギングのベストプラクティス

CI環境では、ビルドやテストのログが問題の早期発見に役立ちます。以下のポイントに注意しましょう。

• ログは標準出力に適切に出すこと（CIツールが捕捉しやすいため）
• ログのボリュームに注意し、重要な部分を分かりやすく出力
• ログファイルをアーティファクトとして保存・共有可能にする
• JSON形式ログを利用し自動解析を容易にする

CircleCIやGitHub Actionsの例

streamlit_app.logをCIのアーティファクトに設定し、テスト失敗時にはログファイルの中身を確認可能にする運用がよく行われます。

12. まとめ

本記事では、Streamlitアプリでのロギングについて、基礎から応用、さらにCI環境での運用まで幅広く解説しました。

• Pythonの標準loggingモジュールを利用した基本ロギング
• カスタムロガー作成による柔軟なログ管理
• 非同期ロギングを実装しパフォーマンス向上
• JSON形式でのログ構造化と一貫性確保
• ロギングがシステムパフォーマンスに与える影響と測定方法
• ログデータ分析・可視化の代表的手法
• 複数モジュール間でのログ統合の重要性
• トラブルシューティングに役立つログ活用例
• 自動テスト・CI環境でのロギングベストプラクティス

ロギングは単なる情報出力だけでなく、アプリの品質を大きく向上させる重要な技術です。ぜひ段階的に導入・活用して、Streamlit開発に役立ててください。

参考リンク

• Python logging公式ドキュメント
• Streamlit公式サイト
• python-json-logger GitHub
• ELKスタックとは – Elastic
• Fluentd 公式サイト
• Grafana Loki 公式サイト
• pytest 公式ドキュメント

最後まで読んでいただきありがとうございます。
ご質問がある方は、コメント欄をご活用ください。