分散処理初心者が実行においてつまづいた３つのこと

この記事は「ウィルゲート Advent Calendar 2023」の 3 日目の記事です。

adventar.org

こんにちは、ウィルゲート開発室の田島です。

現在私は業務で分散処理を扱っているのですが、新しい試みということもあり社内に知見がなく、何をするにも手探りな状態でした。その中で処理を実行するにあたり特につまづいたことを3つ、備忘録がてら共有できればと思います。

なお分散処理フレームワークは Apache Spark を、実行環境には Amazon EMR を採用しております。これらの概要については下記リンク先をご確認ください。

spark.apache.org

aws.amazon.com

Apache Spark はデフォルトで128MBごとに分散処理を行う

Apache Spark では処理対象のデータをデフォルトで128MBごとに分割し、分散処理の実行を行います。

デフォルトではSparkはファイルの各ブロック(HDFSではブロックはデフォルトでは128MB)に1つのパーティションを生成します

mogile.web.fc2.com

私が処理対象としていたテキストファイルは128MBよりも小さいものだったため、このままでは分割されず、単一の処理で実行されてしまいます。128MBよりも小さいデータを対象としたい場合、あるいはより細かい単位で分割したい場合は入力データの読み込み時に明示的に分割数を指定してあげる必要がありました。

// テキストファイルの読み込み時に64分割する
JavaRDD<String> rdd = context.textFile(inputPath, 64);

ローカルのApache Spark のバージョンと Amazon EMR のバージョンを揃える

当たり前の話といえば当たり前の話ですが、Amazon EMR で分散処理を実行する際は、開発に用いたApache Spark のバージョンに対応するバージョンのEMRを指定する必要があります。

こちら、ここ数ヶ月の間でAmazon EMR の UI がより分かりやすく改善されたため、現在はあまり迷う方はいないかもしれません。

s3への書き込み中に `java.io.IOException: File already exists` で落ちる

処理の都合上大量のファイルをs3上に出力していたのですが、実行量が一定を越えると java.io.IOException: File already exists で処理が異常終了する事象が発生しました。

調査の結果Web上でたびたび報告されている事象であること、また根本原因が File already exists ではなく他の部分にあることは分かったのですが、結果的に解決はできず、現状は実行する単位を細かく分けることで回避しています。「実行量が一定を越えると」と表現しましたが、実行時間にあるのか実行量にあるのか、あるいは処理対象に問題があるのかも調査しきれていない状態です。