AWS EMR Serverless Spark应用Executor无法创建/扩展问题求助

核心问题根源 #

你使用parallelize将Driver本地集合转换为RDD的方式是关键问题：parallelize基于Driver内存中的数据创建RDD，默认分区数等于Driver的CPU核数，且Spark会优先在Driver节点执行这类任务，完全无法触发Executor的资源扩展。同时日志中的NPE错误大概率和广播变量的跨节点传输异常有关，进一步加剧了任务无法分发的问题。

1. 替换parallelize为分布式读取逻辑 #

绝对不要先将S3数据加载到Driver内存再parallelize，直接用Spark的分布式数据源读取数据，让Spark自动拆分数据分区并分发到Executor：

错误示例（导致任务在Driver运行）：

// 先读取到Driver本地集合，再转RDD
List<Record> s3Records = readS3RecordsToLocal();
JavaRDD<Record> rdd = sc.parallelize(s3Records);

正确示例（分布式加载，触发Executor调度）：

// 直接从S3读取分布式RDD
JavaRDD<Record> rdd = sc.textFile("s3://your-bucket/path/to/data")
    .map(line -> parseStringToRecord(line));

如果使用Dataset API：

Dataset<Record> dataset = spark.read()
    .format("csv") // 替换为你的数据格式（parquet/json等）
    .option("header", "true")
    .load("s3://your-bucket/path/to/data")
    .map(row -> convertRowToRecord(row), Encoders.bean(Record.class));

2. 调整EMR Serverless Spark配置参数 #

在任务提交或应用配置中添加以下参数，确保动态分配和Executor资源配置正确：

# 开启并配置动态分配
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=2  # 强制启动至少2个Executor，避免仅Driver运行
spark.dynamicAllocation.maxExecutors=50  # 对应200vCPU（每个Executor4vCPU）
spark.dynamicAllocation.initialExecutors=2

# 配置Executor资源规格（匹配EMR Serverless的vCPU/内存比例：1vCPU=2GB内存）
spark.executor.cores=4
spark.executor.memory=8g

# 禁止Driver本地执行，强制任务分发到Executor
spark.driver.localExecution.enabled=false

3. 修复广播变量相关NPE错误 #

日志中的BlockManagerId.executorId()空指针问题，需从序列化和广播数据大小两方面解决：

• 确保广播的Record类实现Serializable接口，或使用Kryo序列化优化：

  spark.serializer=org.apache.spark.serializer.KryoSerializer
  spark.kryo.registerClasses=com.your.package.Record  # 替换为你的类路径
  

• 如果广播数据集过大（超过10GB），放弃广播，改用Spark的Join操作关联两个S3数据集，避免单节点内存压力和传输异常。

4. 验证任务执行状态 #

提交任务后，通过以下方式确认修复效果：

• 查看CloudWatch中EMR Serverless的ExecutorCount指标，确认Executor数量随负载增长；
• 打开EMR Serverless提供的Spark UI链接，检查Stage的任务分布，确认任务在多个Executor上执行。

内容的提问来源于stack exchange，提问作者Shai Barak