Hadoop: プラグ可能なシャッフルとプラグ可能なソート

はじめに

プラグ可能なシャッフルとプラグ可能なソート機能を使用すると、組み込みのシャッフルおよびソートロジックを代替実装に置き換えることができます。これのユースケースの例としては、MapノードからReducerノードへのデータのシャッフルにHTTP以外の別のアプリケーションプロトコル（RDMAなど）を使用することや、ハッシュ集計とLimit-Nクエリを可能にするカスタムアルゴリズムでソートロジックを置き換えることがあります。

重要: プラグ可能なシャッフルとプラグ可能なソート機能は、実験的であり不安定です。つまり、提供されているAPIは、今後のHadoopのバージョンで変更され、互換性が損なわれる可能性があります。

カスタムシャッフルとカスタムソートの実装

カスタムシャッフル実装には、NodeManagerで実行されるorg.apache.hadoop.yarn.server.nodemanager.containermanager.AuxServices.AuxiliaryService実装クラスと、Reducerタスクで実行されるorg.apache.hadoop.mapred.ShuffleConsumerPlugin実装クラスが必要です。

Hadoopが提供するデフォルトの実装を参考として使用できます。

  • org.apache.hadoop.mapred.ShuffleHandler
  • org.apache.hadoop.mapreduce.task.reduce.Shuffle

カスタムソート実装には、Mapperタスクで実行されるorg.apache.hadoop.mapred.MapOutputCollector実装クラスと、(ソート実装によってはオプションで) Reducerタスクで実行されるorg.apache.hadoop.mapred.ShuffleConsumerPlugin実装クラスが必要です。

Hadoopが提供するデフォルトの実装を参考として使用できます。

  • org.apache.hadoop.mapred.MapTask$MapOutputBuffer
  • org.apache.hadoop.mapreduce.task.reduce.Shuffle

設定

シャッフルを提供するNodeManagerで実行される補助サービス（デフォルトではShuffleHandler）を除き、すべてのプラグ可能なコンポーネントはジョブタスクで実行されます。つまり、これらはジョブ単位で構成できます。シャッフルを提供する補助サービスは、NodeManager構成で構成する必要があります。

ジョブ構成プロパティ (ジョブ単位)

プロパティ デフォルト値 説明
mapreduce.job.reduce.shuffle.consumer.plugin.class org.apache.hadoop.mapreduce.task.reduce.Shuffle 使用するShuffleConsumerPlugin実装
mapreduce.job.map.output.collector.class org.apache.hadoop.mapred.MapTask$MapOutputBuffer 使用するMapOutputCollector実装

これらのプロパティは、mapred-site.xmlで設定して、すべてのジョブのデフォルト値を変更することもできます。

コレクタークラス構成では、コンマ区切りのコレクター実装リストを指定できます。この場合、マップタスクは、いずれかの実装が正常に初期化されるまで、各実装を順番にインスタンス化しようとします。これは、特定のコレクター実装が特定のタイプのキーまたは値とのみ互換性がある場合に役立ちます。

NodeManager構成プロパティ、すべてのノードのyarn-site.xml

補助サービスを構成するには、マニフェストファイルを使用する方法と、構成を使用する方法（従来の方法）の2つがあります。マニフェストファイルを使用する場合、補助サービス構成は構成から読み取られません。

マニフェストを使用する場合、yarn-site.xmlでプロパティyarn.nodemanager.aux-services.manifest.enabledをtrueに設定して、機能を有効にする必要があります。ファイルパスは、yarn-site.xmlのyarn.nodemanager.aux-services.manifestプロパティで設定するか、エンドポイントhttp://nm-http-address:port/ws/v1/node/auxiliaryservicesへのPUT呼び出しを介して各NMにファイルを送信できます。ファイルパスが構成に設定されている場合、NMはyarn.nodemanager.aux-services.manifest.reload-msで指定された間隔で、このファイルに新しい変更がないかチェックします（デフォルトは0です。間隔<=0に設定すると、自動的にリロードされません）。

それ以外の場合は、次のプロパティを設定して、構成を介して補助サービスを構成します。

プロパティ デフォルト値 説明
yarn.nodemanager.aux-services ...,mapreduce_shuffle 補助サービス名
yarn.nodemanager.aux-services.mapreduce_shuffle.class org.apache.hadoop.mapred.ShuffleHandler 使用する補助サービスクラス
yarn.nodemanager.aux-services.%s.classpath なし 関連するjarファイルとすべての依存関係のjarファイルを含むローカルディレクトリ。単一のjarファイルを指定するか、/dep/*を使用してdepディレクトリの下にあるすべてのjarをロードできます。
yarn.nodemanager.aux-services.%s.remote-classpath なし jarファイルへのリモートの絶対パスまたは相対パス

HDFSからjarファイルをロードする例

マニフェストの使用

{
  "services": [
    {
      "name": "mapreduce_shuffle",
      "version": "1",
      "configuration": {
        "properties": {
          "class.name": "org.apache.hadoop.mapred.ShuffleHandler"
        }
      }
    },
    {
      "name": "AuxServiceFromHDFS",
      "version": "1",
      "configuration": {
        "properties": {
          "class.name": "org.apache.auxtest.AuxServiceFromHDFS2"
        },
        "files": [
          {
            "src_file": "hdfs:///aux/test/aux-service-hdfs.jar",
            "type": "STATIC"
          }
        ]
      }
    }
  ]
}

または構成の使用

<configuration>
    <property>
        <name>yarn.nodemanager.aux-services</name>
        <value>mapreduce_shuffle,AuxServiceFromHDFS</value>
    </property>

    <property>
        <name>yarn.nodemanager.aux-services.AuxServiceFromHDFS.remote-classpath</name>
        <value>/aux/test/aux-service-hdfs.jar</value>
    </property>

    <property>
        <name>yarn.nodemanager.aux-services.AuxServiceFromHDFS.class</name>
        <value>org.apache.auxtest.AuxServiceFromHDFS2</value>
    </property>
</configuration>

ローカルファイルシステムからjarファイルをロードする例

マニフェストの使用

{
  "services": [
    {
      "name": "mapreduce_shuffle",
      "version": "1",
      "configuration": {
        "properties": {
          "class.name": "org.apache.hadoop.mapred.ShuffleHandler"
        }
      }
    },
    {
      "name": "AuxServiceFromHDFS",
      "version": "1",
      "configuration": {
        "properties": {
          "class.name": "org.apache.auxtest.AuxServiceFromHDFS2"
        },
        "files": [
          {
            "src_file": "file:///aux/test/aux-service-hdfs.jar",
            "type": "STATIC"
          }
        ]
      }
    }
  ]
}

または構成の使用

<configuration>
    <property>
        <name>yarn.nodemanager.aux-services</name>
        <value>mapreduce_shuffle,AuxServiceFromHDFS</value>
    </property>

    <property>
        <name>yarn.nodemanager.aux-services.AuxServiceFromHDFS.classpath</name>
        <value>/aux/test/aux-service-hdfs.jar</value>
    </property>

    <property>
        <name>yarn.nodemanager.aux-services.AuxServiceFromHDFS.class</name>
        <value>org.apache.auxtest.AuxServiceFromHDFS2</value>
    </property>
</configuration>

重要: デフォルトのmapreduce_shuffleサービスに加えて補助サービスを設定する場合は、新しいサービスキーをyarn.nodemanager.aux-servicesプロパティに追加する必要があります（例：mapred.shufflex）。次に、対応するクラスを定義するプロパティはyarn.nodemanager.aux-services.mapreduce_shufflex.classである必要があります。あるいは、補助サービスマニフェストファイルを使用する場合は、サービスをサービスリストに追加する必要があります。