MaterialsEN中文ホームブログ技術ブログ分散処理に入門してみた(Hadoop+Spark)分散処理に入門してみた(Hadoop+Spark)作成日:2014/10/19更新日:2018/07/12技術ブログプログラミングHadoopSpark分散処理こんにちは。RDBやデータウェアハウスの仕事に携わることが多かった筆者は、数年前からたびたび聞こえたビッグデータ分析や機械学習のための分散処理フレームワークに興味を覚えたものの、ついぞアクセスしないままここまで来てしまいました。今回ブログを書くにあたって、せっかくなのでイチから手さぐり入門し、いまさら他人に聞けない分散処理の初歩からhadoop・sparkを触ってみるまでをまとめたいと思います。1.分散処理の基礎知識1-1.分散処理の処理方式:MapReduceまず分散処理とは、ひとつの計算処理をネットワークで接続した複数のコンピュータで同時並列で処理することです。ビッグデータ活用の市場が日々大きくなるに従って、数百テラ~ペタのデータ処理も珍しいものではなくなっており、日常的にこの規模のデータを扱うシステムでは、現実的な時間的・費用的コストで処理する工夫が必要不可欠です。データ処理の高速化のためには、単純に処理サーバを高性能にすることも手段の一つですが、例えばサーバの処理性能を2倍にすると費用的コストは2倍では収まりませんし、将来処理するデータが更に増えた場合に性能を上げるのも非常にコストがかかります。ひとつの手段としては、安価なサーバを複数台用意して処理を割り振ることで、全体の処理能力を向上させることが考えられます。これには処理割り振りのための複雑なプログラミングが必要になりますが、コスト的には安く、将来的な拡張も簡単です。この分散処理について、2004年にGoogleが論文で発表した独自の分散処理フレームワークの処理方式がMapReduceです。GoogleはこのMapReduceを使って、Webサイト内の文字列検索や、特定の単語が含まれるWebページのリスト作成などを行っていると発表しました。これらの処理は一見シンプルではありますが、その扱うデータ量が2004年8月当時すでに3ペタバイト/1カ月という膨大さであったため、非常に驚きをもって受け入れられました。tableの基盤技術を取り入れて実装されたのがHadoopです。Hadoopとしてオープンソース公開されたことで、分散処理を簡単に実装することが可能になり、ビッグデータ活用市場の拡大と分散処理の需要に乗って爆発的に広がりました。現在では様々な有名企業が自社のシステムにHadoopを導入しています。具体的に導入事例をあげてみますと、Facebook広告の解析や「フェイスブック・インサイト」という解析ツール上でサイト来訪ユーザ数やいいね!の数を集計するためHadoopを利用。毎日135TB超のデータを処理。Visaカードの利用パターンを分析することで不正利用を検知するシステムにHadoopを導入。730億件のトランザクションデータ処理を数週間から13分に短縮した。リクルート「Hotpepper」「Suumo」「じゃらん」などの20を超えるWebアプリケーションで、バッチ処理・ログ解析・レコメンドなど多岐にわたってHadoopを導入。...などなど、Webサービス・SNSに携わる企業や、データマイニングで導入例が多いように見受けられます。ログ解析や機械学習を利用したカスタマーへのレコメンド、バッチによる集計処理等に利用されることが多いようです。この傾向は、Web系業界が扱う、膨大かつ将来的に更に拡大していくデータ・ログを分析・解析するのにHadoopが適しているからと言えるのではないでしょうか。1-2-2.Hadoopの特徴では、なぜHadoopがそれらのシステムに向いているのでしょうか。Hadoopがもともと大規模データを扱いやすい分散処理であるからとも言えますが、私は以下のようなHadoopの設計思想と特徴が需要にマッチしていたからだと考えます。数ある特徴の中で、筆者が特に重要では?と思ったものを挙げてみます。1オープンソースの分散処理フレームワークMapper、Reducerと呼ばれる2つのスクリプトを作成するだけで、簡単に分散処理を実装できます。分散処理を独自で開発するには、プロセス起動・監視・プロセス間の通信等の管理、データ処理の効率的な分散、特定ノード故障時の動作など、複雑な考慮が必要不可欠ですが、フレームワークを使用することによりそのほとんどをHadoopに任せることができるようになりました。それらがすべてオープンソース実装されているためソフトウェアの導入コストも低く、運用時に不具合が発生したときにも自分でソースコードを見て修正することも可能です。2安価なサーバをノードとして増やすことで簡単にスケールできる将来的にどんどん増え続けることが見込まれるビッグデータやWebサービスのシステムでは、サーバ性能の拡張が必要になることも多いと思います。冒頭でご説明した通り、分散処理を導入していれば、ひとつの処理サーバの高性能化(=スケールアップ)ではなく、安価なサーバを複数増やす(=スケールアウト)の方式をとることが可能で、コストも安価で済みます。Hadoopではスケールアウトを前提として設計されているため、サーバ台数の増加に比例して性能を上げていくことが可能です。耐障害性を重視して設計されており、処理中に一部の構成ノードが故障しても、すぐに別ノードへ処理を振り分けられます。(ただし、マスターサーバの障害については対処が無く単一障害点になっているそうです。Streamingというツールが用意されており、標準入出力を持つ言語であればあらゆる言語でスクリプトを作成できます。HiveやPigといったDSLも提供されています。5大きなデータを処理するバッチ処理には向くが、リアルタイム処理には向かない分散処理を実行するためにオーバーヘッドが大きく、パフォーマンスが求められるリアルタイム処理には向かない傾向があります。いったデータを扱うシステムでは、RDBとの住み分けや役割分担も視野に入れることも考えられます。上記のようなHadoopの特徴は、以下のような基礎構成要素から成り立ちます。簡単に見ていきましょう。tableに相当しています。Mapreduce分散処理フレームワーク。Mapperとreducerというプログラムを作成するだけでユーザは分散処理を実行することができます。Streamingというツールで、標準入出力があるプログラミング言語であればどんな言語でもMapperとReducerを記述できます。System)分散処理ファイルシステム。1台のサーバのストレージに収まりきらない大容量のデータを、何台かのサーバに分割して配置し管理するための仕組み。HDFSは読み書きを高速化するため、ファイルを一定の大きさのブロック(デフォルトで64メガバイト)に分割し、複数の記憶装置に分散して保存し、読み込み書き込みを並列に実行できるようにしています。・HbaseHDFS上に構築された列指向分散データベース。高い書き込み性能と優れた拡張性を持ち、拡張した後も書き込み性能がそれほど落ちないという特徴があります。MapReduceの内部処理について少しだけ掘り下げます。MapReduceは、Mapフェーズ、Shuffleフェーズ、Reduceフェーズの順に処理されます。ユーザが意識するのはMap・Reduceフェーズのみで、この2フェーズの動作をプログラム実装するだけで分散処理を実現できます。これがMapper、Reducerです。図2のような一連の処理のひと塊りをジョブと呼び、それを下記図3のようにタスクという小さい処理に分割してリソース状況や処理の進行状況、故障の有無によってノードに振り分け処理するのが分散処理です。2014/10月現在、HadoopMapReduceにはVersion1と2が存在し(いわゆるMR1とMR2)、それぞれHadoop1.X.XとHadoop2.X.Xに搭載されています。違いとしては、従来のMR1がジョブを実行するマスター側のジョブトラッカーとスレーブ側のタスクトラッカーで処理されるのに対し、MR2がYARNという汎用的な分散処理ハンドリングのフレームワークでマスター・スレーブを実現していることが挙げられます。詳しくは下記をご参照のこと。Sparkをご紹介します。SparkはHadoopのMapReduce部分に置き換わることを目指して開発された、Scalaで分散処理を行うフレームワークで、いわば高速化されたMapReduceといえます。HadoopではHDFSに貯めたデータに対して機械学習を行い高度な解析を行うシステムがよく導入されていると前述しましたが、機械学習などの繰り返し処理を行うものではなかなか性能が出にくい理由があります。MapReduceで各ノードにジョブを起動する際プログラムの起動に15秒ほどかかったり、ジョブの中で分析対象データをHDFSから取得→集計→HDFSに出力→データを取得...と処理したりすると、ストレージへのオーバーヘッドも無視できない処理時間になります。そこで、繰り返し処理やドリルダウン分析を行うために、HDFSに特殊なキャッシュを設け、インメモリで高速な分散処理を行うことを考えたのがSparkです。数TBあるログデータを分析をしたい時、以下のように比較できます。このようにSparkのインメモリ分散処理ではHDFSへの書き込みが少ない分非常に高速(※10~100倍高速だそうです)で、繰り返しの処理やドリルダウン分析に適しているようです。詳しくは下記参照のこと。http://www.cloudera.co.jp/products-services/cdh/apache-spark.html1-3-1.Sparkの特徴Sparkの特徴も簡単にご紹介します。Datasets)繰り返し利用するデータをキャッシュ上に保持することが可能な仕組み。このため繰り返し処理で非常に高いパフォーマンスを発揮します。通常キャッシュは揮発性ですが、RDDではMapReduceが保持していた高い耐障害性も兼ね備えていて、必要最低限の復旧が可能であるそうです。2DSLMaper・Reducerの代わりにScalaのDSLを記述できる。汎用的にプログラミングできるようになっています。自前では分散ファイルシステムを持っていませんが、HadoopのHDFSを利用可能です。managerで動きますが、前述したHadoop2のYARN上でも稼動します。これにより、HadoopとSpark双方の得意な部分を組み合わせ、バッチをHadoop+オンライン分析をSparkなどの組み合わせたシステムも容易に作ることができます。4インメモリでも高い耐障害性キャッシュ上でデータセットを保持しますが、Hadoop同様高い耐障害性を持っており、たとえRDDのデータが一部損なわれてもRDD内に「入力元となるRDD」と「処理内容」を保持しているためにRDDデータを使って再作成することができます。詳しくは下記参照のこと。http://www.ibm.com/developerworks/jp/opensource/library/os-spark/長くなってしまいましたが、これまでのHadoop・Sparkの情報をまとめてみると...・HadoopとSpark、それぞれの得意分野を組み合わせることで、安価かつ高速で、耐障害性の高い分析システムの構築が可能。・数テラ~数ペタのビッグデータ分析をHadoopバッチジョブとSparkオンライン検索で自在に操ることが可能。いうことが言えそうですが...本当にそんな夢のようなことが可能なのでしょうか?次章、実際に試してみたいと思います。2.HadoopとSparkの実践それでは実際に構築して試してみたいと思います。今回は触り心地を見るためなのであくまで安価に済ますべく、AmazonEMRやCDHなどを使わない素のHadoop(疑似分散モード)をCentOSに構築します。使用する環境とインストールする対象は以下の通りです。...私有のPCにVMware上でインストール。1024MB・jdk1.7.0.65・Hadoop2.5.0・Hive0.13.1・scala2.10.4・Spark1.1.02-1.HadoopのインストールCentOSを立ち上げ、専用ユーザでログインした直後から開始します。ユーザは「hdspark」としました。epel-release-6-8.noarch.rpm//jdkをインストール。/input*/稼働確認がてら、オーソドックスなサンプルwordcountを使ってみます。5//成功!2-2.HIVEのインストールこのままMapperとReducerをプログラミングしてもよいのですが、筆者はRDBに慣れているため、せっかくなのでHIVEを使ってみます。HIVEとは、Facebookによって開発された、RDBを使う感覚でHadoopジョブを実行できる仕組みです。HiveQLというSQLライクのDSL言語を使うだけで内部で勝手にMapとReduceに変換して実行される優れモノです。row(s)hive>//成功!HIVEもインストールできたところで、HDFS上に仮想データベースを作成してみます。システムhttp://www.kasen-suibo.metro.tokyo.jp/im/other/tsim0110g.html00:00,新宿,000:10,新宿,100:20,新宿,3...の形式で、毎日10分ごとの各計測点の降水量が記載されているCSVファイルです。ビッグデータと言えるほどの規模とはいきませんが、これを使って仮想データベースを作成してみます。row(s)//成功!データの読み込みに時間がかかる半面、書き込みは非常に速い印象でした。14400行×約180ファイル=260万件超でしたが、ファイル分割のオーバーヘッド分を含めても10秒ほどで全てアップロード完了しており、HBaseの書き込み性能の高さの一端かもしれません。2-3.Sparkのインストールさて、残るSparkのインストールです。modeで稼働させるため今回はビルドは行わないものとします。141//成功!2-4.Hadoopで出力したデータをSparkで分析してみるそれでは、すべての準備が整ったところでHadoopとSparkでいろいろ触ってみることにします。4//成功!ちゃんとHDFS上のファイルは参照できているようです。これならばHadoopで出力された結果の分析もできるはず!//Hadoopの出力結果を使ってドリルダウン分析//2-2で作成したHIVEのuryouテーブルを使用します。...調べてみると、どうやらHIVEのファイル出力は区切り文字が固定で「^A」になる模様。変更できない、これは若干不便...。出力はできたので先に進んでみます。s//成功!3秒足らずで完了しています。Hadoopはその性質上、ジョブ起動にオーバーヘッド分の時間がありちょっとした集計でも時間がかかっていますが、Sparkはこの実装部分でも頑張っているとのことで、ほぼタイムラグ無しでジョブ実行できるそうです。22倍の差がありました。(もっともHIVEのパフォーマンスは、Javaで記述するMapper・Reducerと比べても変換がある分遅いようなので、純粋にHadoopとの比較とはいきませんが。)3.最後にHadoopとSparkの連携が簡単にでき、特別な意識なく分散処理が行えることがお分かりいただけましたでしょうか。オープンソース実装のためかまだまだ不具合やバグがあり、構築するのに苦労しましたが、商用であればAmazonEMRなどのオールインワンのクラウドサービスも存在し、今後ますます需要が広がるのではと思います。Sparkではまだまだ日本語の記事も少なくノウハウも無いようですが、今後どんどん進化して使えるようになってくれれば面白くなると感じました。拙い記事ですが、HadoopとSparkの凄さのほんの一端を、感じていただければ幸いです。MaterialsEN。