すべてのプロダクト
Search

このプロダクト

    E-MapReduce:ESS (既存ユーザーのみ利用可能)

    ドキュメントセンター

    E-MapReduce:ESS (既存ユーザーのみ利用可能)

    最終更新日:Jun 21, 2026

    EMR Remote Shuffle Service (ESS) は、E-MapReduce (EMR) 向けの拡張機能で、コンピューティングエンジンのシャッフル操作を最適化します。

    背景情報

    従来のシャッフルメカニズムには、以下の課題があります：

    • 大容量データのシナリオでは、シャッフル書き込みによってディスクへのスピルが発生し、書き込み増幅につながる場合があります。

    • シャッフル読み取りでは小さなネットワークパケットが大量に生成され、接続リセットエラーが発生する場合があります。

    • シャッフル読み取りでは多数の小さな I/O リクエストとランダムリードが発生し、ディスクと CPU に大きな負荷がかかります。

    • マッパー (M) とリデューサー (N) の数が数千に達すると、ネットワーク接続の総数 (M × N) により、ジョブが完了できなくなる場合があります。

    • NodeManager と Spark Shuffle Service は同一プロセスで実行されます。シャッフルデータ量が極めて大きい場合、NodeManager が再起動し、YARN スケジューリングの安定性に影響します。

    ESS には、以下のメリットがあります：

    • プル型シャッフルではなくプッシュ型シャッフルを使用することで、マッパーのメモリプレッシャーを低減します。

    • I/O 集約をサポートし、シャッフル読み取りの接続数を M × N から N に削減するとともに、ランダムリードをシーケンシャルリードに置き換えます。

    • 2 レプリカメカニズムをサポートし、フェッチ失敗の発生確率を低減します。

    • コンピュート・ストレージ分離アーキテクチャをサポートし、コンピューティングクラスターから切り離した別のハードウェア環境に Shuffle Service をデプロイできます。

    • Kubernetes 上で Spark を実行する際に、ローカルディスクへの依存を解消します。

    次の図は、ESS のアーキテクチャを示しています。ESS

    制限事項

    このドキュメントは、EMR-3.39.1 より前の EMR バージョン、EMR-4.x 系のバージョン、および EMR-5.5.0 より前のバージョンにのみ適用されます。EMR-3.39.1 以降および EMR-5.5.0 以降については、RSS をご参照ください。

    クラスターの作成

    たとえば EMR-4.5.0 では、ESS を含むクラスターを 2 つの方法で作成できます：

    • E-MapReduce Shuffle Service クラスターを作成します。[Software Configuration] ページで、[Cluster Type] で [Shuffle Service] を選択します。必要なサービスは [ESS (1.0.0)] です。

    • E-MapReduce Hadoop クラスターを作成します。[Software Configuration] ページの [Cluster Type] セクションで、Hadoop、Kafka、Druid などのタイプを選択します。次に、[Cloud Native Options] (例：ECS) と [Product Version] (例：EMR-4.5.0) を設定します。ページに、対応する [required services] (例：HDFS、YARN、Spark) と、バージョン付きの [optional services] (例：ESS、HBase、Flink) が表示されます。

    クラスターの作成方法の詳細については、「Create a cluster」をご参照ください。

    ESS の使用

    Spark で ESS を使用するには、Spark のジョブの投入時に次のパラメータを追加します。パラメータの設定方法の詳細については、「Edit jobs」をご参照ください。

    Spark パラメータの詳細については、「Spark Configuration」をご参照ください。

    パラメータ

    説明

    spark.shuffle.manager

    値は org.apache.spark.shuffle.ess.EssShuffleManager にする必要があります。

    spark.ess.master.address

    アドレスを <ess-master-ip>:<ess-master-port> の形式で指定します。

    パラメータは次のとおりです：

    • <ess-master-ip>：マスターノードの IP アドレス。

    • <ess-master-port>：ポート番号。9097 に設定する必要があります。

    spark.shuffle.service.enabled

    値を false に設定します。

    EMR Remote Shuffle Service を使用するには、デフォルトの外部シャッフルサービスを無効にする必要があります。

    spark.shuffle.useOldFetchProtocol

    値を true に設定します。

    これにより、レガシーシャッフルプロトコルとの互換性が有効になります。

    spark.sql.adaptive.enabled

    値を false に設定します。

    EMR Remote Shuffle Service はアダプティブ実行をサポートしていません。

    spark.sql.adaptive.skewJoin.enabled

    パラメータ

    ESS サービスの設定ページには、ESS のすべてのパラメータが表示されます。

    パラメータ

    説明

    デフォルト

    ess.push.data.replicate

    2 レプリカ機能を有効または無効にします。有効な値：

    • true：2 レプリカ機能が有効になります。

    • false：2 レプリカ機能が無効になります。

    説明

    本番環境では、この機能を有効にすることを推奨します。

    true

    ess.worker.flush.queue.capacity

    ディレクトリあたりのフラッシュバッファ数。

    説明

    パフォーマンスを向上させるために、複数のディスクを設定できます。読み取りと書き込みのスループットを最適化するため、ディスク 1 台あたり 2 つ以下のディレクトリを使用することを推奨します。

    各ディレクトリのフラッシュバッファが消費するヒープメモリは ess.worker.flush.buffer.size * ess.worker.flush.queue.capacity で、256 KB * 512 = 128 MB です。各ディレクトリが提供するスロット数は、ess.worker.flush.queue.capacity パラメータの値の半分です。たとえば合計 28 ディレクトリの場合、総メモリ消費量は 128 MB * 28 = 3.5 GB、総スロット数は 512 * 28 / 2 = 7168 です。

    512

    ess.flush.timeout

    ストレージレイヤーにデータをフラッシュする際のタイムアウト時間。

    240 s

    ess.application.timeout

    アプリケーションのハートビートタイムアウト。この期間内にハートビートを受信しない場合、ESS はアプリケーションのリソースをクリーンアップします。

    240 s

    ess.worker.flush.buffer.size

    フラッシュバッファのサイズ。バッファがこのサイズを超えると、ESS はデータをディスクにフラッシュします。

    256 KB

    ess.metrics.system.enable

    モニタリングを有効または無効にします。有効な値：

    • true：モニタリングを有効にします。

    • false：モニタリングを無効にします。

    false

    ess.worker.offheap.memory

    コアノードのオフヒープメモリサイズ。

    4 GB

    ess.worker.memory

    コアノードのヒープメモリサイズ。

    4 GB

    ess.master.memory

    マスターノードのヒープメモリサイズ。

    4 GB