EMR Serverlessを導入して痛感したこと

うちのチームが月15万円規模のバッチ処理をEMR Serverlessに移行して3ヶ月。正直、最初の1ヶ月は地獄でした。

これまでEC2ベースのEMRクラスタを手で立ち上げていたんですが、スケーリング判断が属人的で、深夜のジョブ失敗対応が月2〜3回。「サーバーレス化すればこの煩雑さから解放される」という甘い幻想を持ってました。

実際に本番化してみて気づいたのは、Serverlessだからこそハマる落とし穴がある、ってこと。自動スケーリング・ワーカー数の予測・CloudWatch連携──どれもドキュメント通りにはいきませんでした。

この記事では、うちが実装した設定とチーム運用で学んだ失敗パターンをまるごと共有します。

EMR Serverlessの基本動作と、うちが見落としたこと

EMR Serverlessはスパークワークロードに特化したマネージドサービスなんですが、スケーリングの意思決定をアプリケーション設計に組み込む必要があります。「スケーリングはAWS側が勝手にやってくれる」という想定は甘かった。

導入前のうちの想定と、実装後の現実をまとめるとこんな感じです：

項目	当初の期待	現実
スケーリング判断	AWS自動で最適化	アプリ設計で制御が必要
コスト・パフォーマンス	自動的に最適化	初期設定で大きく変動
ワーカー数の影響	無視してOK	不適切だと実行時間が3倍に
監視負荷	ゼロ	設計なしでは原因追跡が地獄

実装のポイントはワーカー設定です。これが結構重要なんです：

ApplicationConfiguration:
  Name: "batch-processing-app"
  ReleaseLabel: "emr-7.1.0"  # 2026年時点の最新
  Architecture: "ARM64"  # コスト効率重視
  
  # ワーカー設定──これが結構重要
  DefaultJobRunConfig:
    ExecutionRoleArn: "arn:aws:iam::xxx:role/emrtask"
    
    # Driver設定
    SparkSubmitJobDriver:
      SparkSubmitParameters: |-
        --class com.example.BatchJob
        --conf spark.driver.memory=4g
        --conf spark.driver.cores=2
        --conf spark.executor.memory=6g
        --conf spark.executor.cores=2
        --conf spark.executor.instances=10
        # 重要：動的割り当ての初期値
        --conf spark.dynamicAllocation.enabled=true
        --conf spark.dynamicAllocation.minExecutors=2
        --conf spark.dynamicAllocation.maxExecutors=50
        --conf spark.dynamicAllocation.executorIdleTimeout=60s

この設定、最初は maxExecutors=100 で設定してました。するとデータ量10GBのジョブなのに、ワーカー50個起動してしまって、その月の請求が予想の2倍に。

ここで重要な気付きがあります。EMR Serverlessは起動したインスタンス数ではなく、APU（Amazonの独自計算単位）で課金されます。1ワーカーあたり4APU。50ワーカーだと200APU。1時間あたり約$1.68かかる計算になるんです。

コスト最適化のために実装した3つの工夫

1. ジョブ特性別のプロファイル設定

うちのバッチ処理は3パターンに分かれてます：

• 高速小ジョブ：5GB未満、実行時間5分以内
• 中規模ジョブ：50GB程度、実行時間30分前後
• 大規模ジョブ：1TB超、実行時間2時間以上

これぞれを固定的に maxExecutors 設定するのではなく、CloudFormation / AWS CDKで動的に切り替える構成にしました：

# AWS CDK での EMR Serverless 構成
from aws_cdk import (
    aws_emrserverless as emr,
    aws_iam as iam,
    aws_s3 as s3,
    core
)

class BatchProcessingStack(core.Stack):
    def __init__(self, scope: core.Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)
        
        # S3 バケット（ジョブスクリプト・入出力データ格納）
        bucket = s3.Bucket(
            self, "BatchBucket",
            versioned=True,
            block_public_access=s3.BlockPublicAccess.BLOCK_ALL
        )
        
        # IAM ロール
        job_role = iam.Role(
            self, "EMRJobRole",
            assumed_by=iam.ServicePrincipal("elasticmapreduce.amazonaws.com"),
            inline_policies={
                "S3Access": iam.PolicyDocument(
                    statements=[
                        iam.PolicyStatement(
                            actions=["s3:GetObject", "s3:PutObject"],
                            resources=[bucket.arn_for_objects("*")]
                        )
                    ]
                )
            }
        )
        
        # 小ジョブ用アプリケーション
        app_small = emr.CfnApplication(
            self, "BatchAppSmall",
            release_label="emr-7.1.0",
            type="SPARK",
            architecture="ARM64",
            name="batch-small-jobs",
            auto_start_configuration={"enabled": True},
            auto_stop_configuration={"enabled": True, "idle_timeout_minutes": 5}
        )
        
        # 大ジョブ用アプリケーション
        app_large = emr.CfnApplication(
            self, "BatchAppLarge",
            release_label="emr-7.1.0",
            type="SPARK",
            architecture="ARM64",
            name="batch-large-jobs",
            auto_start_configuration={"enabled": True},
            auto_stop_configuration={"enabled": True, "idle_timeout_minutes": 30}
        )
        
        self.app_small_arn = app_small.attr_arn
        self.app_large_arn = app_large.attr_arn
        self.job_role_arn = job_role.role_arn
        self.bucket_name = bucket.bucket_name

このアプローチで、小ジョブの実行時間が平均5秒短縮。月で約30分のコンピュートタイムが削減されました。地味ですが、積み重ねると大きいんです。

2. 自動スケーリングの監視・チューニング

最初の失敗は「Spark設定に全依存してた」こと。CloudWatchメトリクスを見てなかったので、実行時にワーカーがどう動いているのかが不透明でした。

# CloudWatch メトリクス取得スクリプト
import boto3
from datetime import datetime, timedelta

emr = boto3.client('emr-serverless')
cloudwatch = boto3.client('cloudwatch')

def analyze_job_scaling(app_id: str, job_id: str):
    """ジョブのスケーリング動作を分析"""
    
    # ジョブ情報取得
    job = emr.get_job_run(
        ApplicationId=app_id,
        JobRunId=job_id
    )
    
    start_time = job['jobRun']['createdAt']
    end_time = job['jobRun']['updatedAt']
    
    # CloudWatch メトリクス取得
    response = cloudwatch.get_metric_statistics(
        Namespace='AWS/EMRServerless',
        MetricName='OverallProgress',
        Dimensions=[
            {'Name': 'ApplicationId', 'Value': app_id},
            {'Name': 'JobId', 'Value': job_id}
        ],
        StartTime=start_time,
        EndTime=end_time,
        Period=60,
        Statistics=['Average', 'Maximum']
    )
    
    # ワーカー数の推移を推測（メモリ使用量から逆算）
    memory_metrics = cloudwatch.get_metric_statistics(
        Namespace='AWS/EMRServerless',
        MetricName='TotalAllocatedMemory',
        Dimensions=[
            {'Name': 'ApplicationId', 'Value': app_id}
        ],
        StartTime=start_time - timedelta(minutes=5),
        EndTime=end_time + timedelta(minutes=5),
        Period=30,
        Statistics=['Average']
    )
    
    print(f"Job {job_id} Analysis:")
    print(f"Duration: {(end_time - start_time).total_seconds() / 60:.1f} minutes")
    print(f"Status: {job['jobRun']['state']}")
    
    # メモリ使用量をAPUに換算
    for point in memory_metrics['Datapoints']:
        apu_count = point['Average'] / 4096  # 1APU = 4GB
        print(f"Estimated APU: {apu_count:.1f} at {point['Timestamp']}")

# 使用例
analyze_job_scaling('xxxxxxxx', 'yyyyyyy')

このスクリプトで「なぜこのジョブは1時間かかるのか」がようやく可視化されました。

実装して分かったこと：

• タスク数 > ワーカー数 の場合、スケジューリング待機が多発する
• 初期ワーカー数が2だと、ramp-upのオーバーヘッドで5分損失する
• APU計算が整数単位なので、メモリ6GBと8GBは同じ課金になる

3. EventBridge + Lambda による失敗対応自動化

深夜のバッチが失敗しても、朝まで気づかないパターンがありました。これはつらい。EMR Serverlessのジョブ失敗をEventBridgeでキャッチして、Lambda経由でSlack通知 + 自動リトライ。

# Lambda 関数：ジョブ失敗検知と自動リトライ
import boto3
import json
from datetime import datetime

emr = boto3.client('emr-serverless')
slack = boto3.client('sns')  # SNS経由でSlack

def lambda_handler(event, context):
    detail = event['detail']
    app_id = detail['application_id']
    job_id = detail['job_run_id']
    status = detail['state']
    
    if status != 'FAILED':
        return {'statusCode': 200, 'body': 'Not a failure event'}
    
    # ジョブ情報取得
    job = emr.get_job_run(
        ApplicationId=app_id,
        JobRunId=job_id
    )
    
    failure_reason = job['jobRun'].get('stateDetails', 'Unknown')
    
    # 一時的なエラーなら自動リトライ
    retryable_errors = [
        'SparkDeployFailed',
        'ContainerFailure',
        'OutOfMemory'  # メモリ不足は設定で再実行
    ]
    
    should_retry = any(err in failure_reason for err in retryable_errors)
    
    if should_retry:
        # 前回のジョブ設定を取得
        spark_params = job['jobRun'].get('sparkSubmitParameters', '')
        
        # リトライ用にワーカー数を増やす（OOMの場合）
        if 'OutOfMemory' in failure_reason:
            spark_params = spark_params.replace(
                '--conf spark.executor.instances=10',
                '--conf spark.executor.instances=15'
            )
        
        # リトライ実行
        retry_job = emr.start_job_run(
            ApplicationId=app_id,
            ExecutionRoleArn=job['jobRun']['executionRoleArn'],
            JobDriver={
                'SparkSubmit': {
                    'entryPoint': job['jobRun']['jobDriver']['sparkSubmit']['entryPoint'],
                    'entryPointArguments': job['jobRun']['jobDriver']['sparkSubmit'].get('entryPointArguments', []),
                    'sparkSubmitParameters': spark_params
                }
            },
            Tags={'retry_of': job_id, 'retry_count': str(int(job['jobRun'].get('Tags', {}).get('retry_count', 0)) + 1)}
        )
        
        message = f"✅ Job {job_id} failed ({failure_reason}), auto-retrying as {retry_job['jobRunId']}"
    else:
        message = f"❌ Job {job_id} failed with non-retryable error: {failure_reason}"
    
    # Slack通知
    sns.publish(
        TopicArn='arn:aws:sns:ap-northeast-1:xxxx:batch-alerts',
        Subject=f'EMR Job {job_id} Status Changed',
        Message=message
    )
    
    return {
        'statusCode': 200,
        'body': json.dumps({'action': 'retry' if should_retry else 'alert', 'message': message})
    }

これで深夜バッチの失敗対応が大幅に簡素化されました。本当、これがないと精神的に持ちませんね。

AWS構成図：本番EMR Serverless環境

graph TB
    subgraph "EventDriven"
        EventBridge["EventBridge<br/>Job Failure Detection"]
        Lambda["Lambda<br/>Auto Retry Logic"]
    end
    
    subgraph "DataProcessing"
        S3Input["S3 Input Bucket<br/>Raw Data"]
        EMRSmall["EMR Serverless App<br/>Small Jobs<br/>ARM64"]
        EMRLarge["EMR Serverless App<br/>Large Jobs<br/>ARM64"]
        S3Output["S3 Output Bucket<br/>Processed Data"]
    end
    
    subgraph "Monitoring"
        CloudWatch["CloudWatch<br/>Metrics & Logs"]
        SNS["SNS Topic<br/>Slack Integration"]
        Grafana["Grafana<br/>Dashboard"]
    end
    
    subgraph "Orchestration"
        StepFunctions["Step Functions<br/>Workflow Orchestration"]
        EventSchedule["EventBridge Rules<br/>Cron Trigger"]
    end
    
    StepFunctions -->|Route Small| EMRSmall
    StepFunctions -->|Route Large| EMRLarge
    EventSchedule -->|Trigger| StepFunctions
    
    S3Input -->|Read| EMRSmall
    S3Input -->|Read| EMRLarge
    EMRSmall -->|Write| S3Output
    EMRLarge -->|Write| S3Output
    
    EMRSmall -->|Emit Events| EventBridge
    EMRLarge -->|Emit Events| EventBridge
    EventBridge -->|Detect Failure| Lambda
    Lambda -->|Retry| StepFunctions
    
    EMRSmall -->|Metrics| CloudWatch
    EMRLarge -->|Metrics| CloudWatch
    Lambda -->|Alert| SNS
    CloudWatch -->|Query| Grafana
    SNS -->|Notify| Slack["Slack<br/>Channel"]

この構成で実現してるのは：

• 小ジョブ：起動5秒、実行5分程度。汎用アプリケーション
• 大ジョブ：起動30秒、実行2時間以上。高メモリ、長時間実行用
• 失敗検知：EventBridgeが即座にキャッチ、Lambdaが判断・再実行
• 監視：CloudWatch → Grafana で可視化

コスト削減の実測値

xychart-beta
    title "月間 EMR コスト削減推移（2024年11月〜2026年5月）"
    x-axis [11月, 12月, 1月, 2月, 3月, 4月, 5月]
    y-axis "コスト (万円)" 15 50
    line "EC2ベース (Before)" [48, 50, 49, 51, 50, 52, 51]
    line "Serverless (After)" [40, 35, 32, 28, 25, 24, 22]

導入から5ヶ月で、月間コストを48万円 → 22万円に削減できました。削減率は約54%です。

内訳は：

• ワーカー自動スケーリング: 30万円 → 15万円（50%削減）
• 自動停止機能: 12万円 → 5万円（58%削減）
• コンピュートユニット最適化: 6万円 → 2万円（67%削減）

本当は「Serverlessなら自動的に安くなる」と思ってたけど、設定・監視・最適化なしには効果なし。むしろ下手な設定だと逆に高くなります。体感的には、8割が実装と設定の工夫で、残り2割がAWSの基盤性能です。

運用で直面した3つの大きな失敗

失敗1：長時間ジョブのタイムアウト

3時間超のジョブが、2時間50分で強制終了されました。調べたら、デフォルトタイムアウトが3時間に設定されていて、オーバーヘッドで引っかかってたんです。

# Step Functions での タイムアウト設定
step_functions_job = sfn.Task(
    self, "EMRJob",
    resource=emr_job_arn,
    timeout=cdk.Duration.hours(4),  # 4時間に延長
    heartbeat=cdk.Duration.minutes(5),  # 5分ごとに生存確認
    retry_on_error=['States.TaskFailed'],
    max_attempts=3
)

重要：Step Functions と EMR Serverless のタイムアウトは別です。両方設定する必要があります。この落とし穴、ドキュメントに目立たずにしれっと書いてあるだけなんですよね。

失敗2：ワーカーのJVM設定不足

Javaヒープサイズ設定がなく、GC Stop-the-World で頻繁にハングしてました。Spark 3.2以降では自動チューニングされるはずでしたが、ARM64対応で挙動が変わってたんです。

SparkSubmitParameters: |
  --conf spark.executor.memory=6g
  --conf spark.executor.memoryOverhead=2g  # GC用に明示的に領域確保
  --conf spark.driver.memory=4g
  --conf spark.driver.memoryOverhead=1g
  --conf spark.shuffle.memoryFraction=0.2
  --conf spark.storage.memoryFraction=0.6
  --conf spark.sql.adaptive.enabled=true  # AQE有効化

この設定で、GC時間が平均3分 → 30秒に短縮されました。地味に便利です。

失敗3：S3 Consistency イシュー

EMRがS3に書き込んだ直後に読み込むと、ファイルが見えないバグに直面しました。S3の結果整合性の影響かと思いましたが、実は EMRServerlessのアクティブなコンピュートが複数アクティビティを並列実行していて、バッファリング中だったんです。

# 修正案：S3 出力 → 検証待機 → 次のジョブ開始
from time import sleep
import boto3

def wait_for_s3_consistency(bucket, prefix, expected_count, timeout=300):
    """S3のファイル完全書き込みを待機"""
    s3 = boto3.client('s3')
    start = time.time()
    
    while time.time() - start < timeout:
        response = s3.list_objects_v2(
            Bucket=bucket,
            Prefix=prefix
        )
        
        actual_count = response.get('KeyCount', 0)
        
        if actual_count >= expected_count:
            # ファイルのETag確認で、書き込み完了を二重チェック
            objects = response['Contents']
            if all('ETag' in obj for obj in objects):
                return True
        
        sleep(5)
    
    raise TimeoutError(f"S3 consistency timeout for {bucket}/{prefix}")

# Step Functions で使用
wait_for_s3_consistency(
    bucket='output-bucket',
    prefix='batch-results/',
    expected_count=100
)

運用してわかった、本番化のための5つの条件

本番環境で安定稼働させるには、以下の5点が絶対条件だと感じました：

1. 事前の容量計画：ワーカー数は過去3ヶ月のジョブ実行時間・データサイズから推定する
2. 段階的な移行：小ジョブから本番化し、データ量・複雑度を段階的に増やす
3. CloudWatch ダッシュボード：メトリクス・ログを常時可視化する
4. 失敗リトライの自動化：手動介入が入らない仕組みを最初から作る
5. コスト監視の自動化：月間コスト予測・異常検知アラートを設定する

これらなしで本番化するのは、正直リスクが高い。

まとめ

EMR Serverlessはスケーラビリティと自動化の観点では優れていますが、「設定して終わり」ではなく、ジョブ特性に合わせた継続的な最適化が必須です。うちのチームでは：

• 月額コストを54%削減（48万円 → 22万円）
• バッチ失敗対応時間を80%削減（月2〜3時間 → 月10分程度に自動化）
• ジョブ実行時間を平均15%短縮（最適化後のワーカー設定）

を実現できました。正直、最初の3ヶ月は試行錯誤の連続。でも、この設定・監視基盤ができたら、あとは新しいジョブを追加するたびに効率が上がるようになりました。

特に「ARM64アーキテクチャの採用」「自動リトライの仕組み化」「CloudWatchダッシュボードの可視化」の3つは、本番安定性とコスト削減の鍵になった実感があります。これからEMR Serverless導入を検討してる人には、この3つから始めることをお勧めします。