AIにデータパイプラインを変更させるときのIMPACT_ANALYSIS：392件の事故から学んだ必須チェック

1
# 変更依頼: priority_scoreをpriority_valueにリネーム
2
# analyzer.py（変更箇所）
3
df["priority_value"] = calculate_priority(df)  # priority_score → priority_value
4

5
# しかし以下の5ファイルがpriority_scoreを参照していた
6
# 1. trigger_generator.py
7
# 2. report_generator.py
8
# 3. slack_notifier.py
9
# 4. db_inserter.py（DDLも変更が必要）
10
# 5. regression_test_runner.py

1
# 変更前: resource_countはint
2
resource_count: int = get_resources()
3
if resource_count == 3:  # intの比較はOK
4
    stop_processing()
5

6
# 変更後: AIがfloatに変更（計算の都合で）
7
resource_count: float = get_resources()
8
if resource_count == 3:  # floatとintの比較は予期しない結果になりうる
9
    stop_processing()  # 3.0000000001 != 3 で条件不成立

1
# IMPACT_ANALYSIS_CHECKLIST
2

3
## Stage 1: 変更内容の明確化
4
- [ ] 変更対象のファイル名を列挙
5
- [ ] 変更するカラム名/型/フォーマットを明記
6
- [ ] 変更の目的と期待する効果を記述
7

8
## Stage 2: 影響範囲の特定
9
- [ ] 変更するカラム名でプロジェクト全体をgrep
10
- [ ] 変更するファイル名のimport元をgrep
11
- [ ] Parquet/CSV/DBスキーマへの影響確認
12
- [ ] plistファイルへの影響確認
13

14
## Stage 3: launchdジョブへの影響確認
15
- [ ] 変更ファイルを使用するplistの特定
15 collapsed lines
16
- [ ] 実行順序（依存関係）の確認
17
- [ ] スケジュール時刻への影響確認
18

19
## Stage 4: 型チェック
20
- [ ] mypy --strict の実行
21
- [ ] 型変更がある場合、下流の型整合性を確認
22

23
## Stage 5: 実施順序の策定
24
- [ ] DDL変更（DB側）→ Reader変更 → Writer変更 → Test → Deploy
25
- [ ] ロールバック手順の準備
26

27
## Stage 6: デプロイ後確認
28
- [ ] 5分後: エラーログの確認
29
- [ ] 1時間後: データ出力の正常性確認
30
- [ ] 24時間後: 全パイプラインの通し確認

1
# 影響範囲を自動検出するスクリプト
2
import subprocess
3
from pathlib import Path
4

5

6
def find_impact(
7
    search_term: str,
8
    project_root: str,
9
    file_extensions: list[str] | None = None,
10
) -> list[dict[str, str]]:
11
    """変更対象の文字列を参照しているファイルを特定する。"""
12
    extensions = file_extensions or [".py", ".plist", ".sql", ".md"]
13
    results: list[dict[str, str]] = []
14

15
    for ext in extensions:
25 collapsed lines
16
        cmd = [
17
            "grep", "-rn", search_term,
18
            "--include", f"*{ext}",
19
            project_root,
20
        ]
21
        proc = subprocess.run(cmd, capture_output=True, text=True)
22
        for line in proc.stdout.strip().split("\n"):
23
            if not line:
24
                continue
25
            parts = line.split(":", 2)
26
            if len(parts) >= 3:
27
                results.append({
28
                    "file": parts[0],
29
                    "line": parts[1],
30
                    "content": parts[2].strip(),
31
                })
32

33
    return results
34

35

36
# 使用例
37
impacts = find_impact("priority_score", "/path/to/project")
38
print(f"影響範囲: {len(impacts)}箇所")
39
for impact in impacts:
40
    print(f"  {impact['file']}:{impact['line']} - {impact['content']}")

1
❌ 間違った順序:
2
1. Writerを先に変更 → 新カラム名で書き込み
3
2. Readerはまだ旧カラム名で読もうとする → KeyError
4
3. DB DDLは未変更 → INSERT失敗
5

6
✅ 正しい順序:
7
1. DDL変更（ALTER TABLE ADD COLUMN / RENAME COLUMN）
8
2. Reader変更（新旧両方のカラム名に対応する互換コード）
9
3. Writer変更（新カラム名で書き込み）
10
4. テスト実行（全パイプラインの通し）
11
5. 互換コードの除去（旧カラム名対応を削除）
12
6. デプロイ

1
# Stage 2の互換コード例：新旧カラム名の両方に対応
2
def read_priority_value(df):
3
    """priority_score → priority_value のマイグレーション中の互換関数。"""
4
    if "priority_value" in df.columns:
5
        return df["priority_value"]
6
    if "priority_score" in df.columns:
7
        # 旧カラム名でも動作（移行期間中）
8
        return df["priority_score"]
9
    raise KeyError("priority_value も priority_score も見つかりません")

1
# デプロイ後確認の3段階
2
POST_DEPLOY_CHECKS = {
3
    "5分後": {
4
        "確認項目": [
5
            "stderrにエラーが出ていないか",
6
            "プロセスが正常に起動しているか",
7
            "直近のログにWARNING/ERRORがないか",
8
        ],
9
        "見つかる問題": "構文エラー、インポートエラー、設定ミス",
10
    },
11
    "1時間後": {
12
        "確認項目": [
13
            "出力ファイルが正常に生成されているか",
14
            "DBへのINSERTが成功しているか",
15
            "Slack通知が届いているか",
12 collapsed lines
16
        ],
17
        "見つかる問題": "データ形式不整合、スキーマ不一致、型変換エラー",
18
    },
19
    "24時間後": {
20
        "確認項目": [
21
            "全launchdジョブが正常に実行されたか",
22
            "パイプライン全体のデータフローが正常か",
23
            "日次レポートの数値に異常がないか",
24
        ],
25
        "見つかる問題": "日次バッチの依存関係エラー、データ欠損、集計値の不整合",
26
    },
27
}

1
実例: カラム型変更（int→float）のインシデント
2

3
5分後:  ✅ エラーなし（INSERTは成功）
4
1時間後: ✅ データ出力OK（数値は入っている）
5
24時間後: ❌ 日次集計で異常発見
6
  → 前日のint値と当日のfloat値の比較で
7
     「変化率: +0.0000001%」が大量発生
8
  → 監視アラートが鳴りっぱなし

1
## Critical Rules
2

3
### データパイプライン変更時の必須手順（BLOCKING）
4
以下のファイル/データへの変更時は、IMPACT_ANALYSIS必須:
5
- `.parquet` ファイルのスキーマ変更
6
- `.csv` ファイルのカラム変更
7
- DB DDL変更（CREATE/ALTER TABLE）
8
- API レスポンス形式の変更
9
- `.plist` ファイルの変更
10
- データフォーマットの変更（型変更を含む）
11

12
手順:
13
1. 変更内容を明文化
14
2. grep で影響範囲を特定（変更するカラム名/ファイル名/型名で検索）
15
3. launchdジョブへの影響を確認
3 collapsed lines
16
4. mypy --strict を実行
17
5. 実施順序を策定（DDL → Reader → Writer → Test → Deploy）
18
6. デプロイ後確認（5分/1時間/24時間）

1
【ルール追加前】
2
人間: 「priority_scoreカラムをpriority_valueにリネームしてください」
3
AI:   「analyzer.pyのカラム名を変更しました。完了です」
4
→ 下流5ファイルが壊れる
5

6
【ルール追加後】
7
人間: 「priority_scoreカラムをpriority_valueにリネームしてください」
8
AI:   「データパイプライン変更のため、IMPACT_ANALYSISを作成します。
9

10
       影響範囲の調査結果:
11
       - trigger_generator.py:23行目 — priority_scoreを参照
12
       - report_generator.py:67行目 — priority_scoreを参照
13
       - slack_notifier.py:34行目 — priority_scoreを参照
14
       - db_inserter.py:12行目 — DDLにpriority_scoreカラムあり
15
       - regression_test_runner.py:89行目 — priority_scoreを参照
4 collapsed lines
16

17
       合計5箇所の影響があります。
18
       実施順序: DDL変更 → 互換コード追加 → 全ファイル変更 → テスト → 互換コード除去
19
       進めてよろしいですか？」

1
from datetime import datetime
2

3

4
def generate_impact_analysis(
5
    change_description: str,
6
    changed_files: list[str],
7
    search_terms: list[str],
8
    project_root: str,
9
) -> str:
10
    """IMPACT_ANALYSISのテンプレートを生成する。"""
11
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
12

13
    impacts: list[str] = []
14
    for term in search_terms:
15
        results = find_impact(term, project_root)
42 collapsed lines
16
        for r in results:
17
            impacts.append(f"  - {r['file']}:{r['line']} — {r['content']}")
18

19
    template = f"""# IMPACT_ANALYSIS
20
生成日時: {timestamp}
21

22
## 変更内容
23
{change_description}
24

25
## 変更対象ファイル
26
{chr(10).join(f'- {f}' for f in changed_files)}
27

28
## 影響範囲（自動検出）
29
{chr(10).join(impacts) if impacts else '  検出なし'}
30

31
## チェックリスト
32
- [ ] Stage 1: 変更内容の明確化 — 完了
33
- [ ] Stage 2: 影響範囲の特定 — 上記参照
34
- [ ] Stage 3: launchdジョブへの影響確認
35
- [ ] Stage 4: mypy --strict 実行
36
- [ ] Stage 5: 実施順序の策定
37
- [ ] Stage 6: デプロイ後確認（5分/1時間/24時間）
38

39
## 実施順序
40
1. DDL変更
41
2. Reader変更（互換コード追加）
42
3. Writer変更
43
4. テスト実行
44
5. 互換コード除去
45
6. デプロイ
46
"""
47
    return template
48

49

50
# 使用例
51
analysis = generate_impact_analysis(
52
    change_description="priority_score → priority_value カラムリネーム",
53
    changed_files=["analyzer.py"],
54
    search_terms=["priority_score"],
55
    project_root="/path/to/project",
56
)
57
print(analysis)

1
データパイプライン変更:  平均2.3時間（下流プログラムの特定に時間がかかる）
2
テスト不足:            平均1.1時間（原因特定は比較的容易）
3
デプロイミス:          平均0.8時間（設定ファイルの修正で解決）
4
インポート失敗:        平均0.5時間（エラーメッセージが明確）
5
launchd設定:          平均0.3時間（plutil -lint で特定可能）

1
デプロイ直後（5分以内）:  40%
2
1時間以内:              25%
3
24時間以内:             20%
4
24時間超:               15%（日次バッチで初めて顕在化）