ESS（EMR Remote Shuffle Service）是EMR在优化计算引擎的Shuffle操作上，推出的扩展组件。

背景信息

目前Shuffle方案缺点如下：
  • Shuffle Write在大数据量场景下会溢出，导致写放大。
  • Shuffle Read过程中有大量的网络小包导致Connection reset问题。
  • Shuffle Read过程中存在大量小数据量的IO请求和随机读，对磁盘和CPU造成高负载。
  • 对于M*N次的连接数，在M和N数千的规模下，作业基本无法完成。
  • NodeManager和Spark Shuffle Service是同一进程，当Shuffle的数据量特别大时，通常会导致NodeManager重启，从而影响YARN调度的稳定性。
EMR推出的基于Shuffle的ESS服务，可以优化目前Shuffle方案的问题。ESS优势如下：
  • 使用Push-Style Shuffle代替Pull-Style，减少Mapper的内存压力。
  • 支持IO聚合，Shuffle Read的连接数从M*N降到N，同时更改随机读为顺序读。
  • 支持两副本机制，降低Fetch Fail概率。
  • 支持计算与存储分离架构，可以部署Shuffle Service至特殊硬件环境中，与计算集群分离。
  • 解决Spark on Kubernetes时对本地磁盘的依赖。
ESS设计架构图如下。ESS

创建集群

以EMR-4.5.0版本为例，您可以通过以下两种方式创建ESS的集群：
  • 创建E-MapReduce的Shuffle Service集群。SHUFFLE
  • 创建E-MapReduce的Hadoop集群。ESS

集群创建详情请参见创建集群。

使用ESS

Spark使用ESS时，只需在提交Spark作业时添加如下两个配置项，配置详情请参见作业编辑。
配置项 描述
spark.shuffle.manager 固定值org.apache.spark.shuffle.ess.EssShuffleManager。
spark.ess.master.address 填写格式<ess-master-ip>:<ess-master-port>。
涉及参数如下：
  • <ess-master-ip>：Master节点的公网IP地址。
  • <ess-master-port>：固定值9097。

配置项说明

您可以在ESS服务配置页面，查看ESS所有的配置项。
配置项 描述 默认值
ess.push.data.replicate 是否开启两副本。取值包含：
  • true：开启两副本。
  • false：不开启两副本。
说明 建议生产环境开启两副本。
true
ess.worker.flush.queue.capacity 每个目录的Flush buffer数量。
说明 为了提升性能，您可以配置多块磁盘。为了提升整体的读写吞吐量，建议一块磁盘不多于2个目录。

每个目录的Flush buffer所消耗堆内的内存为ess.worker.flush.buffer.size * ess.worker.flush.queue.capacity，即256 KB * 512 = 128 MB。每个目录提供的槽位（slots）数量是该参数的一半。例如，总共28个目录，则整体内存消耗是128 MB * 28 = 3.5 GB，整体的slots数量是512 * 28 / 2 = 7168。

512
ess.flush.timeout Flush到存储层的超时时间。 240s
ess.application.timeout Application心跳超时时间，超时会清理Application相关资源。 240s
ess.worker.flush.buffer.size Flush buffer大小，超过最大值会触发刷盘。 256k
ess.metrics.system.enable 是否打开监控。取值包含：
  • true：打开监控。
  • false：不打开监控。
 false
ess_worker_offheap_memory Worker堆外内存大小。 4g
ess_worker_memory Worker堆内内存大小。 4g
ess_master_memory Master堆内内存大小。 4g