AWS Lambda durable functionsで並列操作mapをためしてみた

def map(
    inputs: Sequence[U],
    func: Callable,
    name: str | None = None,
    config: MapConfig | None = None,
) -> BatchResult[T]

def process_item(
    context: DurableContext,
    item: U,
    index: int,
    items: Sequence[U]
) -> T:
    # 処理ロジック
    return result

from aws_durable_execution_sdk_python.config import (
    MapConfig,
    CompletionConfig,
    ItemBatcher,
)

config = MapConfig(
    max_concurrency=10,  # 同時処理数
    item_batcher=ItemBatcher(max_items_per_batch=5),  # バッチサイズ
    completion_config=CompletionConfig.all_successful(),  # 成功基準
)

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)
from aws_durable_execution_sdk_python.config import MapConfig, CompletionConfig

@durable_execution
def lambda_handler(event: dict, context: DurableContext) -> dict:
    """失敗許容の設定例"""

    product_ids = event.get("product_ids", ["P001", "P002", "P003"])

    def check_inventory(ctx: DurableContext, product_id: str, index: int, items: list) -> dict:
        def call_api(_):
            return {"product_id": product_id, "stock": 100}
        return ctx.step(call_api, name=f"check_{product_id}")

    # 80%以上成功すればOK（大量処理で一部失敗を許容する場合）
    config = MapConfig(
        max_concurrency=5,
        completion_config=CompletionConfig(tolerated_failure_percentage=20),
    )

    results: BatchResult[dict] = context.map(product_ids, check_inventory, config=config)

    # 結果を表示
    for res in results.all:
        print(res)

    return {
        "success_count": results.success_count,
        "failure_count": results.failure_count,
    }

アイテム: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
max_concurrency=3 の場合:

step1: [1] [2] [3]  ← 3つ同時実行
step2: [4] [5] [6]  ← 1〜3が終わったら次の3つ
step3: [7] [8] [9]
step4: [10]

アイテム: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
ItemBatcher(max_items_per_batch=5) の場合:

バッチ1: [1, 2, 3, 4, 5]  ← 5個まとめて1回の関数呼び出し
バッチ2: [6, 7, 8, 9, 10] ← 残り5個をまとめて1回

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)
from aws_durable_execution_sdk_python.config import MapConfig, ItemBatcher, BatchedInput

@durable_execution
def lambda_handler(event: dict, context: DurableContext) -> dict:
    """item_batcherの使用例"""

    item_ids = ["item1", "item2", "item3", "item4", "item5",
                "item6", "item7", "item8", "item9", "item10"]

    def process_batch(ctx: DurableContext, item: BatchedInput[None, str], index: int, items: list) -> list:
        def call_api(_):
            return [{"id": item, "processed": True}]
        return ctx.step(call_api, name=f"batch_{index}")

    # 5個ずつバッチ化して処理
    config = MapConfig(
    max_concurrency=2,
    item_batcher=ItemBatcher(max_items_per_batch=5),
    )

    # 戻り値はバッチ内アイテム数と同数の結果リストを返す
    results: BatchResult[list] = context.map(item_ids, process_batch, config=config)

    # 結果を表示
    for res in results.all:
        print(res)

    return {"processed_count": results.success_count}

アイテム: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

Step1 バッチ化: [[1,3,,5,7,9], [2,4,6,8,10]]  ← 2つのバッチに分割
Step2 同時実行: 2つのバッチを同時に処理      ← max_concurrency=2

1. ユーザー情報を取得（1回）
2. そのユーザーの注文一覧を取得 → 注文IDリスト [O001, O002, O003]
3. 各注文の詳細を並列取得（map）
4. 全注文に含まれる商品IDを抽出 → 商品IDリスト [P001, P002, P003, P004]
5. 各商品の在庫を並列確認（map）
6. 結果を統合して返却

import time

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)
from aws_durable_execution_sdk_python.config import MapConfig

@durable_execution
def lambda_handler(event: dict, context: DurableContext) -> dict:
    """1ユーザーの注文詳細と在庫状況を取得"""

    user_id = event.get("user_id", "U001")

    # 1. ユーザー情報を取得
    def fetch_user(_):
        time.sleep(1)
        return {"user_id": user_id, "name": "Taro Yamada", "email": "taro@example.com"}

    user = context.step(fetch_user, name="fetch_user")

    # 2. 注文一覧を取得（注文IDのリストを返す）
    def fetch_order_list(_):
        time.sleep(1)
        return ["O001", "O002", "O003"]  # 動的な注文IDリスト

    order_ids = context.step(fetch_order_list, name="fetch_order_list")

    # 3. 各注文の詳細を並列取得（map）
    def fetch_order_detail(ctx: DurableContext, order_id: str, index: int, items: list) -> dict:
        def call_api(_):
            time.sleep(2)  # 各APIは2秒かかる
            # 注文詳細のモックデータ
            return {
                "order_id": order_id,
                "items": [f"P00{index * 2 + 1}", f"P00{index * 2 + 2}"],
                "total": 1000 * (index + 1),
            }
        return ctx.step(call_api, name=f"fetch_order_{order_id}")

    order_results: BatchResult[dict] = context.map(order_ids, fetch_order_detail)

    # 4. 全注文から商品IDを抽出
    product_ids = []
    for order_result in order_results.all:
        if order_result.result:
            product_ids.extend(order_result.result["items"])
    product_ids = list(set(product_ids))  # 重複除去

    # 5. 各商品の在庫を並列確認（map）
    def check_inventory(ctx: DurableContext, product_id: str, index: int, items: list) -> dict:
        def call_api(_):
            time.sleep(1)  # 各APIは1秒かかる
            return {"product_id": product_id, "stock": 100 - index * 10, "available": True}
        return ctx.step(call_api, name=f"check_inventory_{product_id}")

    # 同時実行数を制限（在庫APIの負荷を考慮）
    inventory_config = MapConfig(max_concurrency=3)
    inventory_results: BatchResult[dict] = context.map(
        product_ids, check_inventory, config=inventory_config
    )

    # 6. 結果を統合して返却
    return {
        "user": user,
        "orders": [r.result for r in order_results.all if r.result],
        "inventory": [r.result for r in inventory_results.all if r.result],
        "summary": {
            "order_count": order_results.success_count,
            "product_count": inventory_results.success_count,
        },
    }

順次実行の場合:
  ユーザー情報(1秒) + 注文一覧(1秒) + 注文詳細×3(6秒) + 在庫確認×6(6秒) = 14秒

map()による並列実行の場合:
  ユーザー情報(1秒) + 注文一覧(1秒) + 注文詳細(2秒) + 在庫確認(2秒) = 6秒
  ※在庫確認はmax_concurrency=3なので、6件を2回に分けて実行

# parallel()向き：異なる処理を同時実行
context.parallel([
    fetch_user,       # ユーザー情報API
    fetch_orders,     # 注文一覧API
    fetch_inventory,  # 在庫一覧API
])

# map()向き：同じ処理を複数アイテムに適用
order_ids = ["O001", "O002", "O003"]  # 動的なリスト

def fetch_order_detail(ctx, order_id, index, items):
    return ctx.step(lambda _: call_order_api(order_id), name=f"fetch_{order_id}")

context.map(order_ids, fetch_order_detail)

@durable_execution
def lambda_handler(event: dict, context: DurableContext) -> dict:
    """parallel()とmap()を組み合わせた例"""

    # Step 1: parallel()で異なる情報を同時取得
    def fetch_user(ctx):
        return ctx.step(lambda _: call_user_api(), name="fetch_user")

    def fetch_order_ids(ctx):
        return ctx.step(lambda _: call_order_list_api(), name="fetch_order_ids")

    initial_data = context.parallel([fetch_user, fetch_order_ids])

    user = initial_data.all[0].result
    order_ids = initial_data.all[1].result

    # Step 2: map()で注文詳細を並列取得
    def fetch_order_detail(ctx, order_id, index, items):
        return ctx.step(lambda _: call_order_detail_api(order_id), name=f"order_{order_id}")

    orders = context.map(order_ids, fetch_order_detail)

    return {"user": user, "orders": [r.result for r in orders.all]}