現在JavaScriptの設定が無効になっています。このページを利用するにはJavaScriptの設定を有効にしてください。トップ全記事一覧テクノロジー2010.05.27Hadoopを使いこなす(3)こんにちは、R&D統括本部の吉田です。前回と前々回の記事では、Hadoopのカスタマイズポイントを解説しました。単純な問題に対しては、map関数とreduce関数の組み合わせだけでも、処理を行うことができますが、ある程度複雑な問題に対しては、今までに解説したカスタマイズポイントの活用が重要になるという話でした。今回は実際のサービスの事例を紹介し、カスタマイズポイントがどう活用されているか紹介したいと思います。ABYSSABYSSの事例を紹介しましょう。ABYSSとは、検索サービスを簡単に構築できる社内プラットフォームのことで、詳しくは以下のTechBlog記事で紹介しています。先日無事に社内リリースされました。新検索プラットフォーム「ABYSS」これがABYSSのすべてだ!!ABYSSでは、各種検索データのストレージにHadoopを使い、様々な処理を行っています。Hadoopで行われているMapReduce処理の一つに「検索差分データのマージ処理」があります。今回は、この処理について解説したいと思います。「検索差分データ」とは、例えばWEBページの検索インデックスに対して、後からこのURLのデータを更新したい、とか、このURLを削除したい、といった差分を記述したデータのことです。以下のような形式になっています。http://cu.yahoo.co.jp/1行目は、「http://latlonglab.yahoo.co.jp/」の内容の更新し、2行目は追加、3行目は削除する差分データです。差分データは、検索インデックスに対して後からデータを追加したり削除したりします。検索インデックスを新規に作成する場合などは、検索対象となる元データをまず作成し、そこから検索インデックスを生成します。検索対象となる元データを作成し、新規に検索インデックスを生成する方法と、差分データを用意して、後から検索インデックスを更新する2通りの方法があるわけです。差分更新は、検索インデックスに対して直接反映されるため、元データに対しては変更が反映されません。ABYSSでは、元データを後から解析処理などに使えるように、元データに対しても差分データをマージしています。その部分が今回解説するMapReduce処理です。「検索の元データ」とは、例えば以下のようなデータです。Transitional//EN"...また、元データがどういう形式で記述され、どのように検索ができるのかという情報は、スキーマファイルに記述されています。/></Schema>元データは、Url、Title、Bodyの順に並んでいて、ユニークキーがUrlである、という情報を記述しています。差分データをマージする処理は、一見単純な処理に思われるかもしれません。ABYSSでは、次のような制約があります。検索インデックスは複数あり、数千規模まで増える可能性がある検索インデックスごとに、別々のスキーマファイルがあり、元データの形式が異なる検索インデックスごとにマージ処理の入力も出力先も異なる多くの別々の検索サービスを統合するのがABYSSの役割なので、検索インデックスは複数あり、数千規模まで増える可能性があります。検索インデックス1つに対して1回のMapReduceを行うと、数千回MapReduceを行う可能性があり、非常に効率が悪いです。複数の検索インデックスの元データに対するマージ処理を、1回のMapReduceで行う必要があります。検索インデックスごとにデータの形式が異なり、マージ処理の入力も出力先も異なります。いったある程度複雑な問題に対しては、単にmap関数、reduce関数だけでなく、以前の記事で紹介したカスタマイズポイントを上手く利用する必要があります。処理の概要まずは、処理の概要を説明します。上図のように、差分データと、元データをMapReduceの入力ファイルとして読み込みます。MultipleInputsを使い、差分データと、元データには別々のMapperを割り当てます。Mapperでは、スキーマファイルの情報に基づきユニークキーを抽出します。差分データのMapperでは、それに加え、差分データを元データの形式に整形します。Mapperの出力Key/Valueを以下のように、いくつかの要素の複合形式にします。元データの行>ポジションとは、InputSplit内で、その行がどの位置にあるかという数値です。処理の内容とは、差分データの場合、その行を挿入するか(insert)、更新するか(update)、削除するか(insert)、元データの場合は無条件にallとします。次にPartitionerを使い、同じインデックスのデータは、必ず同じReduceTaskで処理されるようにします。GroupingComparatorを使い、インデックス名+ユニークキーでグルーピングします。Comparatorを使い、ファイルの更新時刻の逆順、更新時刻が同じだった場合はポジションの逆順でソートします。結果として、Reducerには、同じインデックスで同じユニークキーのデータがグルーピングされて、新しい順に入力されます。Reducerでは、最新の行のみを出力します。最新の行の処理の内容がremoveだった場合には、出力しません。差分データと元データの中で、同じユニークキーの最新の行を出力するというわけです。最新の行がremoveだった場合はそのユニークキーは削除し、出力しません。結果として、元データに差分データがマージされるわけです。Reducerと同じ処理をMapTask内のCombinerでも行い、転送量を削減しています。MapReduceの出力は、MultipleTextOutputFormatを使い、各インデックスごと別々のフォルダ、ファイルに出力します。処理の詳細を説明していきます。処理の詳細はじめに、解説するカスタムクラスのまとめを簡単に書きますので、わかりにくくなったら参照して下さい。MDPathFilter元データと差分データの両方ともそろっている入力ファイルのみ許可MDTextInputFormat小さいファイルをまとめて一つのInputSplitで扱えるようにMDLineRecordReaderファイルの更新時刻とインデックス名をJobConfにセットMDMapRunnerスキーマファイルから情報を抽出MDDiffMapper差分データのMapperMDAllMapper元データのMapperMDKeyWritableMapの出力Key。新しい順にソートMDValueWritableMapの出力ValueMDHashPartitionerインデックスのハッシュ値で分割MDGroupingComparatorインデックス名+ユニークキーでグルーピングMDReducer最新の行のみを出力。実際に処理するファイルを決定する仕組みです。元データと差分データの両方ともそろっているインデックスのファイルに対してのみ、処理を行うためにPathFilterを活用しています。PathFilterでそうしたファイルのフィルタリングを行い、リソースを有効活用しています。MultipleInputsMultipleInputsは、ファイルごとに別々にMapperと、InputFormatを割り当てる仕組みです。詳細は省きますが、InputSplit自体に、MapperとInputFormatの情報を含めてしまうことで実現しています。以下のように、差分データには、MDDiffMapper、元データにはMDAllMapperを割り当てています。MDDiffMapper.class);InputFormatInputFormatは、CombineFileInputFormatを継承したクラスを使用しています。通常は、少なくともファイル1つに1個のInputSplitが生成されるため、小さいファイルがたくさんあると、MapTaskが膨大な数になってしまい非効率です。CombineFileInputFormatを使えば、小さなファイルをまとめて1つのInputSplitを生成することができます。}}getRecordReader関数では、CombineFileRecordReaderを返します。CombineFileRecordReaderは、ファイルをまとめたInputSplit(CombineFileSplit)を扱うことができ、ファイルごとにオリジナルのRecordReader(MDLineRecordReader)がインスタンス化され、呼び出されます。RecordReaderMapperでは、ファイルの更新時刻と、インデックス名をファイルのパスから取得する必要があります。Mapperの初期化処理だけでそれらの情報を取得することはできません。CombineFileRecordReaderでは、InputSplitに含まれるファイルごとに、オリジナルのRecordReaderがインスタンス化され、初期化処理が呼ばれるようになっています。Mapperから読み出せるようにしています。JobConfに文字列などをセットする方法や、RecordReaderとは何かについては、前の記事を参照して下さい。MapRunnerMapperでは、ユニークキーがどこにあるかという情報や、差分データを元データの形式に変換するための情報を、スキーマファイルから読み出す必要があります。スキーマファイルは、インデックス一つに対して、1回読み出せば十分です。Mapperのインスタンス変数にセットしています。MapRunnerについては前回の記事を参照して下さい。MapperMapperは、差分データを処理するためのMDDiffMapperクラスと、元データを処理するためのMDAllMapperクラスに分けられます。MDDiffMapper、MDAllMapper共に、MapRunnerでスキーマファイルから読み出した情報を元に、ユニークキーを抽出します。MDDiffMapperでは、さらに、差分データを元データの形式に変換します。Mapperの出力Key/Valueは以下の通りです。元データの行>KeyとValueはそれぞれ、MDKeyWritable、MDValueWritableという独自のWritableクラスを使用しています。Writableクラスのカスタマイズ方法は前回の記事を参照して下さい。Partitionerインデックス名のハッシュ値で、Partitionに分割し、ReduceTaskに割り当てます。同じインデックスのファイルは同じReduceTaskで処理されるというわけです。同じインデックスのデータは同じReduceTaskで処理されるので、MapReduceの結果のマージされたファイルは、一つにまとまって出力されるという効果があります。}}Partitionerについては、前回の記事を参照して下さい。ソートComparatorをカスタマイズし、ファイルの更新時刻の逆順、更新時刻が同じだった場合はポジションの逆順でソートするようにしています。これで、新しい順にデータがReducerに渡されることになります。バイト列でそのまま比較する最適化を行っています。Comparatorについては、前回の記事を参照して下さい。グルーピングComparatorをカスタマイズし、同じインデックス名かつ同じユニークキーをグルーピングして、Reducerに渡すようにしています。MapReduceの実行の前処理で、独自のComparatorクラスをGroupingComparatorとして登録します。job.setOutputValueGroupingComparator(MDGroupingComparator.class);ソートの時と同様にバイト列でそのまま比較する最適化を行っています。(Combiner)Reducerには、同じインデックスで同じユニークキーのデータがグルーピングされて、新しい順に入力されます。概要で述べた通り、Reducerでは、最新の行のみを出力します。最新の行の処理の内容がremoveだった場合には、出力しません。差分データと元データの中で、同じユニークキーの最新の行を出力することで、元データに差分データがマージされるわけです。出力のKey/Valueは以下の通りです。nullKeyに実際に出力する行に加えて、インデックス名を指定し、後のOutputFormatでの出力先を決定します。CombinerもReducerと同じ処理を行い、転送量を削減しています。Combinerについては、前回の記事を参照して下さい。OutputFormatOutputFormatには、MultipleTextOutputFormatを継承したクラスを使用することで、インデックスごとに別々のファイルに出力します。}}generateFileNameForKeyValue関数で、Reducerの出力Keyから、インデックス名を抽出し、どのファイルに書き出すかを決定します。generateActualKey関数で、実際にファイルに書き出すKeyを決定します。まとめこうして、差分データが元データにマージされて、インデックスごとにファイルが出力されました。カスタマイズポイントが活用されている雰囲気を感じていただければ幸いです。実際には、もう少し複雑になっていて、説明のために単純化して書いた部分があります。詳細は省きますが、差分データに単純な追加/更新/削除以外の処理があり、それに対応するためにファイルをキャッシュする仕組みを導入したりしています。ABYSSでは、HDFSのファイルをRESTで操作できるようにするなど、Hadoop自体のカスタマイズも行っています。おわりに3回に渡り、Hadoopのカスタマイズポイントを解説し、それらが実際の業務でどう活用されているかを紹介しました。最近では、PigやHive、HadoopStreamingなど、Hadoopでの分散処理を記述する労力を減らす方法がたくさんあります。map関数とreduce関数の組み合わせでできるような処理は、Javaで記述するより、それらの手段を活用した方が効率的です。Hadoopの内部の仕組みを理解し、解説したようなカスタマイズポイントをいじる必要が出てきます。詳しく知りたい場合は、オライリーから出ている象本を読んでみることをおすすめします。Hadoopに特許の利用を許可したそうです。JAPANでは情報技術を駆使して人々や社会の課題を一緒に解決していける方を募集しています。詳しくは採用情報をご覧ください。JAPANのサービス開発を支える技術やデザイン、文化を発信しています。Reserved.このページの先頭へ。