Apache Spark 2.0ジョブは完了するまでに長い時間がかかります

現象

Apache Spark 2.x を使用すると、Spark ジョブがすべて完了しているにもかかわらず、プログラムがまだ実行されているという現象が発生することがあります。たとえば、Spark SQL を使用していくつかの SQL を実行すると、最終的に大量のファイルが生成されます。次に、この SQL のすべての Spark ジョブが実際に完了まで実行されているが、このクエリ ステートメントはまだ実行中であることがわかります。ログから、ドライバーノードがタスクによって生成されたファイルを 1 つずつ最終テーブルのディレクトリに移動していることがわかります。この現象は、ジョブが大量のファイルを生成する場合に発生しやすくなります。この記事では、この問題を解決する方法を紹介します。

なぜこのような現象が起こるのでしょうか?

Spark 2.x は Hadoop 2.x を使用します。生成されたファイルを HDFS に保存すると、次のように、FileOutputCommitter を使用する saveAsHadoopFile が最終的に呼び出されます。

問題は、Hadoop 2.x の FileOutputCommitter 実装にあります。FileOutputCommitter には、commitTask と commitJob という注目すべき 2 つのメソッドがあります。 Hadoop 2.x の FileOutputCommitter 実装では、mapreduce.fileoutputcommitter.algorithm.version パラメータによって commitTask と commitJob の動作が制御されます。具体的なコードは次のとおりです (便宜上、無関係な記述は削除しました。完全なコードは FileOutputCommitter.java にあります)。

ご覧のとおり、commitTask メソッドには、条件判断 algorithmVersion == 1 があります。これは、mapreduce.fileoutputcommitter.algorithm.version パラメータの値で、デフォルトは 1 です。このパラメータが 1 の場合、Task が完了すると、Task によって一時的に生成されたデータは、タスクの対応するディレクトリに移動され、その後、commitJob が呼び出されたときに最終ジョブ出力ディレクトリに移動されます。Hadoop 2.x でのこのパラメータのデフォルト値は 1 です。このため、ジョブは完了しているように見えますが、プログラムはまだデータを移動しているため、ジョブ全体が完了していません。最終的に、commitJob 関数は Spark のドライバーによって実行されるため、実行が遅くなる理由があります。

また、 mapreduce.fileoutputcommitter.algorithm.versionパラメータの値を 2 に設定すると、commitTask の実行時に mergePaths メソッドが呼び出され、タスクによって生成されたデータがタスクの一時ディレクトリからプログラムによって最終的に生成されたディレクトリに直接移動されることがわかります。 commitJob を実行する場合、データを直接移動する必要がないため、当然デフォルト値よりもはるかに高速になります。

Hadoop 2.7.0 より前のバージョンでは、プログラムがこの値を 2 に制限しないため、mapreduce.fileoutputcommitter.algorithm.version パラメータを 1 以外の値に設定することでこれを実現できることに注意してください。ただし、Hadoop 2.7.0 以降では、mapreduce.fileoutputcommitter.algorithm.version パラメータの値は 1 または 2 にする必要があります。詳細については、MAPREDUCE-4815 を参照してください。

Sparkでこのパラメータを設定する方法

問題が見つかりました。プログラムで解決できます。いくつかの方法があります:

• conf/spark-defaults.confでspark.hadoop.mapreduce.fileoutputcommitter.algorithm.version 2直接設定します。これはグローバルに影響します。
• Spark プログラムで直接設定します (spark.conf.set("mapreduce.fileoutputcommitter.algorithm.version", "2"))。これはジョブ レベルです。
• Dataset API を使用して HDFS にデータを書き込む場合は、 dataset.write.option("mapreduce.fileoutputcommitter.algorithm.version", "2") を設定できます。

ただし、Hadoop バージョンが 3.x の場合、mapreduce.fileoutputcommitter.algorithm.version パラメータのデフォルト値はすでに 2 に設定されています。詳細については、MAPREDUCE-6336 および MAPREDUCE-6406 を参照してください。

このパラメータはパフォーマンスに多少の影響を与えるため、Spark 2.2.0 では、このパラメータは Spark 構成ドキュメントconfiguration.htmlに記録されています。詳細については、SPARK-20107 を参照してください。

要約する

以上、Apache Spark 2.0についてご紹介しました。お役に立てれば幸いです。