[Python] ijsonライブラリで大容量JSON（7GB）をストリーミング処理してみた

[project]
dependencies = [
    "ijson>=3.4.0",
]

import ijson
import urllib.request

def extract_metadata_only(url: str, timeout: int = 60) -> dict[str, str | None]:
    """Extract only version and publicationDate from AWS Price List JSON.

    早期終了により、数GBのファイルから数KB分のみ読み取って処理完了。
    """
    with urllib.request.urlopen(url, timeout=timeout) as response:
        version = None
        publication_date = None

        # イベントストリーム形式でパース
        parser = ijson.parse(response)

        for prefix, event, value in parser:
            # トップレベルのversionフィールドを検出
            if prefix == "version" and event == "string":
                version = value
            # トップレベルのpublicationDateフィールドを検出
            elif prefix == "publicationDate" and event == "string":
                publication_date = value

            # 両方見つかったら即座に終了（重要！）
            if version and publication_date:
                break

        return {
            "version": version,
            "publicationDate": publication_date,
        }

def stream_extract_instance_types(url: str, timeout: int = 900) -> dict[str, Any]:
    """Extract instance types from AWS Price List JSON using streaming parser.

    メモリに全体を展開せず、必要な情報のみを逐次抽出。
    """
    with urllib.request.urlopen(url, timeout=timeout) as response:
        instance_types = set()
        version = None
        publication_date = None
        total_products = 0
        instance_products = 0

        parser = ijson.parse(response)
        current_product_family = None

        for prefix, event, value in parser:
            # メタデータ取得
            if prefix == "version" and event == "string":
                version = value
            elif prefix == "publicationDate" and event == "string":
                publication_date = value

            # products配下の処理
            elif prefix.startswith("products."):
                parts = prefix.split(".")

                # products.{SKU} の終了（1製品完了）
                if len(parts) == 2 and event == "end_map":
                    total_products += 1
                    current_product_family = None

                # products.{SKU}.productFamily の値
                elif len(parts) == 3 and parts[2] == "productFamily" and event == "string":
                    current_product_family = value
                    # "Compute Instance" または "Compute Instance (bare metal)"
                    if value.startswith("Compute Instance"):
                        instance_products += 1

                # products.{SKU}.attributes.instanceType の値
                elif (
                    len(parts) == 4
                    and parts[2] == "attributes"
                    and parts[3] == "instanceType"
                    and event == "string"
                ):
                    # Compute Instanceファミリーの場合のみ追加
                    if current_product_family and current_product_family.startswith("Compute Instance"):
                        instance_types.add(value)

        return {
            "instanceTypes": sorted(instance_types),
            "version": version,
            "publicationDate": publication_date,
            "totalProducts": total_products,
            "instanceProducts": instance_products,
        }

def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
    """Lambda handler for EC2 pricing data extraction."""
    url = "https://pricing.us-east-1.amazonaws.com/offers/v1.0/aws/AmazonEC2/current/index.json"

    # ステップ1: メタデータのみ抽出（軽量・高速）
    metadata = extract_metadata_only(url)
    version = metadata["version"]

    # バージョンチェック: 既に処理済みならスキップ
    if check_version_exists(s3_client, bucket, version):
        return {"skipped": True, "version": version}

    # ステップ2: フルストリーミング処理（必要な場合のみ）
    extracted = stream_extract_instance_types(url)
    instance_types = extracted["instanceTypes"]

    # S3へ保存
    save_full_list(s3_client, bucket, version, instance_types)

    return {
        "success": True,
        "version": version,
        "totalCount": len(instance_types)
    }

# ❌ メモリ効率が悪い
with urllib.request.urlopen(url) as response:
    data = response.read().decode("utf-8")  # 約7GBをメモリに展開
    price_list = json.loads(data)           # さらにPythonオブジェクトに展開

# ✅ メモリ効率的
with urllib.request.urlopen(url) as response:
    parser = ijson.parse(response)  # ストリーミング処理
    # 逐次的にイベントを処理