dwh_auditor.analyzer — DWH コスト分析・セキュリティ診断ロジック

dwh_auditor.analyzer パッケージは、Extractor から受け取った Pydantic モデルと config.yaml のしきい値を突き合わせ、診断を行う 純粋な Python ロジック です。

注釈

このパッケージは google.cloud.bigquery を 一切インポートしません。 外部 API 通信が存在しないため、ユニットテストはミリ秒単位で完了します。 ダミーの QueryJob / TableStorage オブジェクトを渡すだけでテストできます。

コスト分析 (analyzer.cost)

高コストクエリの分析ロジック.

注意: このモジュールは google.cloud.bigquery を一切インポートしてはなりません。 純粋な Python ロジックのみで構成し、単体テストがミリ秒単位で完了するようにします。

dwh_auditor.analyzer.cost.analyze_cost(jobs, config)[ソース]

クエリジョブを解析し、高コストクエリのランキングを返す.

キャッシュヒットしたジョブは課金ゼロなので除外します。

パラメータ:

  • jobs (list[QueryJob]) -- QueryJob のリスト (Extractor から受け取る)

  • config (AppConfig) -- AppConfig オブジェクト

戻り値:

CostInsight のリスト (コスト降順、上位 N 件)

戻り値の型:

list[CostInsight]

定常コスト分析 (analyzer.recurring)

定常実行クエリの分析ロジック.

注意: このモジュールは google.cloud.bigquery を一切インポートしてはなりません。

dwh_auditor.analyzer.recurring.analyze_recurring_cost(raw_stats, config)[ソース]

定常実行されているクエリのメタデータを RecurringCostInsight のリストにマップ・計算する.

パラメータ:

  • raw_stats (list[dict[str, Any]]) -- Extractor から抽出した定常クエリの集計辞書

  • config (AppConfig) -- しきい値やコスト単価設定

戻り値:

RecurringCostInsight のリスト

戻り値の型:

list[RecurringCostInsight]

フルスキャン検知 (analyzer.scan)

フルスキャン（非効率クエリ）の検知ロジック.

注意: このモジュールは google.cloud.bigquery を一切インポートしてはなりません。 純粋な Python ロジックのみで構成し、単体テストがミリ秒単位で完了するようにします。

dwh_auditor.analyzer.scan.detect_full_scans(jobs, tables, config)[ソース]

フルスキャンの可能性があるクエリを検知する.

バイト比率バリデーション方式: クエリの課金バイト数が参照テーブルの物理サイズの 90% 以上ならフルスキャンとみなす。

パラメータ:

  • jobs (list[QueryJob]) -- Extractor から抽出したパース対象のジョブ

  • tables (list[TableStorage]) -- テーブルサイズ情報をルックアップするためのリスト

  • config (AppConfig) -- しきい値

戻り値:

FullScanInsight のリスト

戻り値の型:

list[FullScanInsight]

ゾンビテーブル検知 (analyzer.zombie)

テーブルプロファイリング・およびゾンビ（未使用）判定ロジック.

注意: このモジュールは google.cloud.bigquery を一切インポートしてはなりません。 純粋な Python ロジックのみで構成し、単体テストがミリ秒単位で完了するようにします。

dwh_auditor.analyzer.zombie.analyze_table_usage(tables, usage_stats, config, now=None)[ソース]

各テーブルのプロファイル（利用状況とゾンビ判定結果）を返す.

Extractor から受け取った軽量な usage_stats と tables を結合します。

パラメータ:

  • tables (list[TableStorage]) -- TableStorage のリスト (Extractor から受け取る)

  • usage_stats (dict[str, dict[str, Any]]) -- "project.dataset.table" をキーとする統計辞書 (Extractor から受け取る)

  • config (AppConfig) -- AppConfig オブジェクト

  • now (datetime | None) -- 基準となる現在日時 (テスト用)

戻り値:

TableUsageProfile のリスト (ストレージサイズ降順)

戻り値の型:

list[TableUsageProfile]

分析ランナー (analyzer.runner)

分析ランナー: 各 Analyzer を呼び出して AuditResult に集約する.

注意: このモジュールは google.cloud.bigquery を一切インポートしてはなりません。

dwh_auditor.analyzer.runner.run_analysis(top_cost_jobs, heavy_scan_jobs, recurring_stats, table_usages, tables, config, analyzed_days, project_id)[ソース]

全 Analyzer を実行し、総合監査結果を返す.

パラメータ:

  • top_cost_jobs (list[QueryJob]) -- 高コストランキング用のジョブ抽出結果

  • heavy_scan_jobs (list[QueryJob]) -- フルスキャン検知用のジョブ抽出結果

  • recurring_stats (list[dict[str, Any]]) -- 定常実行クエリの抽出結果 (dict)

  • table_usages (dict[str, dict[str, Any]]) -- テーブルごとの利用統計 (dict)

  • tables (list[TableStorage]) -- テーブルストレージ情報

  • config (AppConfig) -- AppConfig オブジェクト

  • analyzed_days (int) -- 分析対象期間 (日数)

  • project_id (str) -- 分析対象の自 GCP プロジェクト ID

戻り値:

AuditResult オブジェクト

戻り値の型:

AuditResult