EMR on EC2 Step提交Spark Flink作业与MWAA集成指南

Amazon EMR产品形态与适用场景

Amazon EMR作为AWS托管的大数据处理服务，提供三种部署形态以满足不同业务需求：EMR Serverless适合无需管理基础设施的场景；EMR on EC2提供最全面的大数据生态支持，包括Spark、Flink、Trino、HBase、Hive等；EMR on EKS则为Kubernetes技术栈用户提供原生集成能力。

从架构设计角度，EMR on EC2支持两种运行模式：

• 瞬态集群：适用于批处理ETL、大规模数据回溯等有明确生命周期的任务，作业完成后自动终止集群
• 长期运行集群：适用于Flink实时计算、Trino即席查询等持续性业务，可配合弹性伸缩保持成本效益

Step API运行机制深度解析

理解Step API的底层机制对于正确使用至关重要。当通过Step提交作业时，EMR会在Master节点启动common-runner.jar进程，该进程负责调用本地的spark-submit或flink run命令。

以下是通过AWS CLI提交Flink作业的示例：

aws emr add-steps --cluster-id j-XXXXXXXXXXXXX \
    --steps Type=CUSTOM_JAR,Name="Flink Job",ActionOnFailure=CONTINUE,\
Jar=command-runner.jar,Args=[flink,run-application,-t,yarn-application,-d,s3://bucket/app.jar]

关键认知：Step状态仅反映common-runner.jar执行命令的结果，而非底层作业的实际运行状态。例如，使用flink run-application模式时，只要作业成功提交到YARN并分配到JobManager容器，命令即返回成功，Step状态变为Completed，但Flink作业可能仍在运行或后续失败。

Spark作业提交策略与内存规划

客户端模式选择

Spark通过spark.yarn.submit.waitAppCompletion参数控制客户端行为：

• true（默认）：客户端进程持续运行直到作业完成，Step状态能准确反映作业结果
• false：提交完成后客户端立即退出，需通过YARN API获取真实状态

Master节点内存容量规划

实践经验表明，每个并行的Step作业（包含spark-submit客户端和common-runner.jar进程）约占用1.4GB内存，建议按2GB/作业进行容量规划。计算公式：

所需Master内存 = 并行作业数 × 2GB + 其他服务预留内存

对于10个并行作业的场景，Master节点至少需要20GB以上可用内存。

高并发场景的状态追踪方案

当设置waitAppCompletion=false时，可通过YARN REST API获取作业状态：

# 按作业名称查询（需确保名称唯一）
curl http://master_ip:8088/ws/v1/cluster/apps?applicationTypes=SPARK&name=my-spark-job

# 按Application ID精确查询
curl http://master_ip:8088/ws/v1/cluster/apps/application_1234567890123_0001

Flink作业提交最佳实践

Flink官方推荐使用Application Mode进行生产环境部署，该模式具有以下优势：

• JobManager在YARN容器中运行，实现资源隔离
• 客户端仅负责提交，不持有长连接
• 配合-d（detached）参数可立即释放客户端资源

flink run-application -t yarn-application -d \
    -Dyarn.application.name="production-flink-job" \
    s3://bucket/flink-app.jar

架构建议：避免使用Per-Job模式的attached mode，该模式会导致客户端进程长期驻留Master节点，在高并发场景下极易耗尽内存资源。

与MWAA工作流集成架构

Amazon Managed Workflows for Apache Airflow（MWAA）提供了与EMR Step的原生集成能力，通过EmrAddStepsOperator可实现作业编排：

from airflow.providers.amazon.aws.operators.emr import EmrAddStepsOperator

add_step = EmrAddStepsOperator(
    task_id='submit_spark_job',
    job_flow_id='j-XXXXXXXXXXXXX',
    steps=[{
        'Name': 'Spark ETL Job',
        'ActionOnFailure': 'CONTINUE',
        'HadoopJarStep': {
            'Jar': 'command-runner.jar',
            'Args': ['spark-submit', '--class', 'com.example.Main', 's3://bucket/app.jar']
        }
    }]
)

通过解析Step执行日志可获取对应的Application ID，进而实现端到端的作业状态追踪与告警机制。