YAML宣言型の意図検証：期待出力をコードではなく設定で定義する

1
$ python pipeline.py && echo "SUCCESS"
2
SUCCESS

1
# job_a.py の末尾
2
output = Path("/data/output_a.parquet")
3
if not output.exists():
4
    raise FileNotFoundError(f"出力がありません: {output}")
5
if output.stat().st_size < 1000:
6
    raise ValueError(f"出力が小さすぎます: {output.stat().st_size} bytes")

1
ジョブのコード（what to do）  ←  分離  →  期待出力の定義（what to expect）

1
import yaml
2
from pathlib import Path
3
from datetime import datetime, timedelta
4
import pyarrow.parquet as pq
5
import csv
6

7
def validate_job(job_name: str, config_path: str) -> list[str]:
8
    """ジョブの出力を検証し、失敗した項目のリストを返す"""
9
    with open(config_path) as f:
10
        config = yaml.safe_load(f)
11

12
    job = config["jobs"].get(job_name)
13
    if not job:
14
        return [f"未定義のジョブ: {job_name}"]
15

47 collapsed lines
16
    errors = []
17
    today = datetime.now().strftime("%Y-%m-%d")
18

19
    for output in job["outputs"]:
20
        path = Path(output["path"].replace("{date}", today))
21
        checks = output["checks"]
22

23
        # 存在チェック
24
        if checks.get("exists") and not path.exists():
25
            errors.append(f"ファイルが存在しない: {path}")
26
            continue  # 以降のチェックはスキップ
27

28
        # サイズチェック
29
        if "min_size_bytes" in checks:
30
            size = path.stat().st_size
31
            if size < checks["min_size_bytes"]:
32
                errors.append(
33
                    f"サイズ不足: {path} ({size} < {checks['min_size_bytes']} bytes)"
34
                )
35

36
        # 鮮度チェック
37
        if "max_age_hours" in checks:
38
            mtime = datetime.fromtimestamp(path.stat().st_mtime)
39
            age_hours = (datetime.now() - mtime).total_seconds() / 3600
40
            if age_hours > checks["max_age_hours"]:
41
                errors.append(
42
                    f"データが古い: {path} ({age_hours:.1f}h > {checks['max_age_hours']}h)"
43
                )
44

45
        # 行数チェック（Parquet）
46
        if checks.get("format") == "parquet" and "min_rows" in checks:
47
            table = pq.read_table(path)
48
            if len(table) < checks["min_rows"]:
49
                errors.append(
50
                    f"行数不足: {path} ({len(table)} < {checks['min_rows']})"
51
                )
52

53
        # カラムチェック（CSV）
54
        if checks.get("format") == "csv" and "required_columns" in checks:
55
            with open(path) as csvfile:
56
                reader = csv.reader(csvfile)
57
                headers = next(reader)
58
                missing = set(checks["required_columns"]) - set(headers)
59
                if missing:
60
                    errors.append(f"カラム不足: {path} — {missing}")
61

62
    return errors

1
  new_job:
2
    description: "新しいジョブの説明"
3
    outputs:
4
      - path: "/data/new_output.parquet"
5
        checks:
6
          exists: true
7
          min_size_bytes: 1000
8
          max_age_hours: 25

1
# CI設定（例）
2
steps:
3
  - name: Validate pipeline outputs
4
    run: python validate_pipeline.py --all --fail-on-error

1
# 閾値の自動算出（初回セットアップ時に実行）
2
import statistics
3

4
sizes = [f.stat().st_size for f in Path("/data/daily/").glob("*.parquet")]
5
mean = statistics.mean(sizes)
6
stdev = statistics.stdev(sizes)
7
suggested_min = int(mean - 2 * stdev)
8
print(f"推奨 min_size_bytes: {suggested_min}")