私は現場でAWS CloudWatch に出力されるエラーログの定期的な確認と集計を担当しています。
手動でのログ確認は手間がかかるので、作業を効率化するためにCloudWatch Logs InsightsでエクスポートしたJSONを元に、エラー内容と発生箇所(path)ごとにログをグループ化し、件数を集計する Pythonスクリプトを作成しました。
ログ分析を効率化したい方の参考になれば幸いです!
準備
まずは CloudWatch Logs Insights を使って、必要なログを検索します。
ロググループを選択し、クエリを実行した後に、右上の 「結果をエクスポート」ボタンをクリック → 「テーブルをダウンロード(JSON)」を選択 すると、検索結果を JSON 形式でエクスポートできます。
保存されたファイルは通常 logs-insights-results.json というファイル名になります。
目的
今回はAWS CloudWatch Logs Insights からエクスポートしたログファイル(JSON形式)を対象に、以下の3つの処理を行うPythonスクリプトを紹介します。
・エラーメッセージとpathの組み合わせでグルーピング
・各グループごとに件数をカウント
・不正な(不完全な)ログはスキップして内容を確認
Pythonスクリプト
import json
from collections import defaultdict
# JSONファイルを開く
with open("logs-insights-results.json", "r", encoding="utf-8") as f:
data = json.load(f)
# data はリスト形式
results = data
# エラーの種類と path でグループ化
group_counts = defaultdict(int)
for entry in results:
try:
message = entry["@message"]
context = message.get("context", {})
detail = context.get("detail", {})
error_message = detail.get("message")
error_path_raw = detail.get("path")
# message や path が無効の場合はスキップ
if not error_message or not isinstance(error_path_raw, list):
raise ValueError("message または path が無効")
error_path = tuple(error_path_raw)
group_key = (error_message, error_path)
group_counts[group_key] += 1
except Exception as e:
# timestamp 付きでスキップログを出力
timestamp_str = entry.get("@timestamp", None)
if timestamp_str:
formatted_timestamp = timestamp_str
else:
formatted_timestamp = "timestampなし"
print(f"スキップされたログ(理由: {e} / timestamp: {formatted_timestamp})")
continue
# 件数の多い順にソートして出力
sorted_group_counts = sorted(group_counts.items(), key=lambda x: x[1], reverse=True)
for (msg, path), count in sorted_group_counts:
print(f"[{count}件] {msg} | path: {path}")
スクリプトの実行方法(初心者向け)
Pythonの初心者でも実行できるように、手順を簡単にまとめました。
1.Pythonのインストール
公式サイトからPythonをインストールしてください
→ https://www.python.org/
インストール時に 「Add Python to PATH」 にチェックを入れてください!
「Add Python to PATH」にチェックを入れずにPythonをインストールすると、コマンドライン(ターミナルやPowerShell)から python が使えなくなります。
2.スクリプトとJSONファイルを準備
上記のPythonコードを group_logs.py という名前で保存します。
logs-insights-results.json を同じフォルダに置きます。
3.コマンドラインで実行
cd スクリプトのあるフォルダ
python group_logs.py
結果がコンソールに表示されます。
実行結果
スキップされたログ(理由: message または path が無効 / timestamp: 2025-06-21 03:38:48.707)
[15件] User not found | path: ('user', 'profile')
[8件] Invalid token | path: ('auth', 'validate')
[3件] Database timeout | path: ('orders', 'submit')
不完全なログは「スキップされたログ」として表示され、内容とタイムスタンプが記録されます。
スキップされてしまったログはタイムスタンプをもとに、あとから手動で確認できます。
おわりに
CloudWatch のエラーログを手動で確認・集計するのは、件数が多くなるととても大変です。
しかし、このような簡易集計スクリプトを作っておくと、エラーの発生件数や新規エラーの検出が格段に楽になります。
小さな改善ですが、日々の運用を効率化したい方の参考になれば幸いです!