分散処理こんにちは、弊社で1ヶ月半ほどインターンとして働いているt_sakaiです。インターンシップの課題として、メモリ使用量が大きくて将来問題になりそうなバッチ処理をスケールアウトできるように書き換えるという課題に取り組みました。Sparkを使ってみたので、本記事ではこれについて書こうと思います。本記事で使っている言語はScalaです。前編ではSparkを選んだ理由とSparkの基礎について説明したので、後編では既存プログラムをSpark用に書き換える際の問題と解決方法について説明します。したかった課題なぜSparkを選んだのかRDDについて簡単なプログラムをSparkで書き換える後編(本記事)本番プログラムをSpark用に書き換えるはまりどころ本番プログラムをSpark用に書き換えるモデル(DB)まわり既存のプログラムを書き換えるときにもっとも大変な作業は、データの入出力まわりです。RDBMSやNoSQLなどのビッグデータ用でないデータソースからのデータ読み込みは大変です。普通のプログラムにおいてはORMや独自モデルが使われていて、コントローラにデータが渡ってきた段階で必要なデータ整形処理などが完了しています。ここで注意しないといけないのは、モデルからデータを受け取る段階で1台のサーバのメモリ上にデータをまるまる置いていることです。Sparkで扱いたいデータというのは基本的に1台のサーバのメモリにおける大きさではありません。そのため、ORMや既存のモデルを使うことは基本的にはできなくなります。(そもそもSparkでは、入力データはHDFSやS3にある程度整形された状態で置いてあることを想定されているような気がします。)同様に出力データが膨大な場合にも、1台のサーバのメモリにデータを集約してからDBやファイルに書き出すことはできません。いずれの場合も、Spark用のデータコネクタや関数を使って入出力を行うことになります。RDDへの読み込みS3などに保存されたCSV形式のファイルこの場合は以下のように読み込めるためかなり簡単です。sc.textFile('s3n://bucketname/Filename')MySQLJdbcRDDというDBコネクタがあり、それを使って読み込むのが簡単なようです。普通にクエリを書くこともできます。こちらのQiitaの記事で具体的な使い方が説明されているのでご覧ください。Connectorというものを使います。ここに書いてあるパラメータを指定すれば、クエリも投げられるようです。RDDからの書き込みファイルへの書き出しファイルへの書き出しは非常に簡単で、以下のメソッドを呼び出すだけです。rdd.saveAsTextFile('s3n://bucketname/Filename')DBへの書き出し(データが小さい場合)もし出力データがそれほど大きくなく、1台のマシンのメモリに乗り切るくらいなのであれば、collectメソッドなどを用いて一度マスターサーバにデータを集約してから書き出すと簡単です。outputは分散されていない普通のデータなので、ORMなどで自由に扱える。DBへの書き出し(データが大きい場合)もし出力データが1台のマシンのメモリに乗り切らない場合は少々大変です。この場合公式のドキュメントでも触れられているように、RDDのforEachPartition関数というものを使ってそれぞれのスレーブサーバでDBコネクションを取るのが定石のようです。}}はまりどころここからは、実際に業務でSparkを使う人向けのかなりディープな話になります。Serializableでないクラスをスレーブで使うと、実行時エラーになる既存のクラスを再利用してSparkスレーブ内で使おうとすると、以下のようなエラーに出くわすことがあります。yourClassSparkスレーブでコードが実行されるとき、オブジェクトやクラスはマスターノードでシリアライズされてスレーブノードへ送られます。スレーブノードで受け取ったデータをデシリアライズして実際のインスタンスを作ります。Sparkスレーブ上で利用される全てのクラスはSerializableである必要があります。Overflowに投稿されたQ&Aしかし、普通に実装したクラスだとSerializableになっていないケースがあります。そのような場合、以下のようにしてSerializableにできる可能性があります。...}このエラーはSparkをローカルモードで動かしたときには起こらないので、結構気付きにくいです。(私はこれに丸3日くらいはまってました。上で紹介したような分かりやすいエラーメッセージが出ないこともありますので、Sparkスレーブでクラスをインスタンス化する際は気をつけてください。スレーブで設定ファイルなどを読み込めない以下のコードのようにローカルファイル読み込みをスレーブ上で行う場合は、そのファイルをスレーブにも渡してやる必要があります。con.save(each)}これはマスターにだけコードをデプロイしてるときに起こりがちなミスです。jarファイルはマスターにだけデプロイしておけば事足りるので気づくのが遅れがちになります。基本的に、マスターとスレーブでフォルダ構成は全く同じにしておくことをお勧めします。エラーメッセージが分かりにくい既に少し触れましたが、Sparkがランタイムエラーで落ちたときのエラーメッセージは非常に不親切です。FAILED_TO_UNCOMPRESS(5)とかいうググりにくいエラーが出たりします。この問題に関してはSparkのJIRA上でも度々議論に上がりますが、なかなか解決されていません。私もこれという解決策は見出せませんでした。経験則としてはメモリを増やすと解決することが多かったです。強制終了するとゴミファイルが残る開発時はSparkプログラムをCtrl+Cで強制終了したりすることが多いのですが、そうするとゴミファイルがスレーブ上に残ってしまう場合が結構あります。私はこれのせいでスレーブサーバのディスクスペースを使い切ってしまって、謎のエラーに出くわしたりしました。この時は要らないファイルを手動で消して対応しました。場合、/tmp/か$SPARK_HOME/work/にゴミファイルが残っていますので、適宜そのファイルを消すようにすればよいと思われます。evaluation(遅延評価)なので、デバッグがやりにくく、メモリ消費量を予測するのも難しいです。並列処理の中でprintデバッグとかをしようとしてもコンソールには出力されず、それぞれのスレーブ上の指定のstdoutファイルに出力されてしまいます。この件についてはまだこれといったベストプラクティスを見つけられていません。詳しい方がいらしゃったら、コメントしていただけると幸いです。--一応、デバッグや性能チューニングについて書かれた公式のスライドもありました。Sparkを使って、メモリ使用量が大きいバッチ処理をスケールアウト初めまして、弊社で1ヶ月半ほどインターンとして働いているt_s...2014-10-14PHPのジェネレータを使おうこんにちは。最近、都内を離れると行くとこ...2014-07-29MongoDBとHiveでごにょごにょ事始めこんにちは、s_mamedaifukuです。Are青山のアドテク会社。アドネットワークやDSPを開発・運用。サーバを止めてみたよ。インターン:実業務体験編インターンでCTR予測をしましたSearchArchiveスターをつけました引用をストックしましたストック一覧を見る閉じる引用するにはまずログインしてくださいログイン閉じる引用をストックできませんでした。再度お試しください閉じる限定公開記事のため引用できません。