【Hadoop】MapReduceを書いてみる

前回はwindowsでHadoopのサンプルアプリ（wordcount）を

環境構築して動かしてみました。

今回は実際にMapReduceをプログラムしてみようと思います。

ここでは、おなじみの形態素解析を行い、文章の品詞を

カウントするという処理を作ってみたいと思います。

1 環境
2 Mavenプロジェクト作成
3 MapReduceプログラム作成
4 実行する
5 まとめ

環境

実施環境としては、

・windows10

・Eclipse

・Maven

・java　1.8

・kuromoji　形態素解析エンジン

を使って行います。

また、対象の文章はこちらもおなじみ桃太郎の一節から。

昔々、あるところに、おじいさんとおばあさんが住んでいました。
おじいさんは山へ芝刈りに、おばあさんは川へ洗濯に行きました。
おばあさんが川で洗濯をしていると、ドンブラコ、ドンブラコと、大きな桃が流れてきました。

昔々、あるところに、おじいさんとおばあさんが住んでいました。

おじいさんは山へ芝刈りに、おばあさんは川へ洗濯に行きました。

おばあさんが川で洗濯をしていると、ドンブラコ、ドンブラコと、大きな桃が流れてきました。

という文章のテキストファイルを用いることとします。

Mavenプロジェクト作成

EclipseにてMavenプロジェクトを作成します。

また、pom.xmlにhadoop関連と、kuromojiの依存関係を追加します。

  <dependencies>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-core</artifactId>
      <version>0.20.2</version>
     </dependency>

    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-common</artifactId>
      <version>3.2.1</version>
    </dependency>

    <dependency>
      <groupId>com.atilika.kuromoji</groupId>
      <artifactId>kuromoji-ipadic</artifactId>
      <version>0.9.0</version>
    </dependency>

  </dependencies>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-core</artifactId>

</dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-common</artifactId>

</dependency>

<groupId>com.atilika.kuromoji</groupId>

<artifactId>kuromoji-ipadic</artifactId>

</dependency>

</dependencies>

※ここでhadoop-coreのversionを0.20.2としているのは、

それ以降のversionでは、Eclipseでjavaアプリケーションを素で実行すると、

hadoop側のプログラムでjava.io.FileクラスのsetReadable()メソッドなど、

フォルダのパーミッションを変更する処理があるのですが、

こちらがjava側の仕様でWindowsでは動作しません。

そのため、このパーミッション変更処理がないversionで動かす必要があり、

0.20.2としています。

MapReduceプログラム作成

プログラム全文を以下のように記述します。

import java.io.IOException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.atilika.kuromoji.ipadic.Token;
import com.atilika.kuromoji.ipadic.Tokenizer;

public class WordCountSample {

	public static class SampleMapper extends Mapper<Object, Text, Text, IntWritable> {

		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

			Tokenizer tokenizer = new Tokenizer();
			List<Token> tokens = tokenizer.tokenize(value.toString());

			for (Token token : tokens) {

				word.set(token.getPartOfSpeechLevel1());
				context.write(word, one);
			}
		}
	}

	public static class SampleReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

		private IntWritable result = new IntWritable();

		public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

			int sum = 0;

			for (IntWritable val : values) {
				sum += val.get();
			}

			result.set(sum);
			context.write(key, result);

		}
	}

	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();
		Job job = new Job(conf, "word count sample");
		job.setJarByClass(WordCountSample.class);
		job.setMapperClass(SampleMapper.class);
		job.setCombinerClass(SampleReducer.class);
		job.setReducerClass(SampleReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		System.exit(job.waitForCompletion(false) ? 0 : 1);
	}

}

import java.io.IOException;

import java.util.List;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.atilika.kuromoji.ipadic.Token;

import com.atilika.kuromoji.ipadic.Tokenizer;

public class WordCountSample {

public static class SampleMapper extends Mapper<Object, Text, Text, IntWritable> {

private final static IntWritable one = new IntWritable(1);

private Text word = new Text();

public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

Tokenizer tokenizer = new Tokenizer();

List<Token> tokens = tokenizer.tokenize(value.toString());

for (Token token : tokens) {

word.set(token.getPartOfSpeechLevel1());

context.write(word, one);

}

public static class SampleReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

private IntWritable result = new IntWritable();

public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

int sum = 0;

for (IntWritable val : values) {

sum += val.get();

}

result.set(sum);

context.write(key, result);

}

public static void main(String[] args) throws Exception {

Configuration conf = new Configuration();

Job job = new Job(conf, "word count sample");

job.setJarByClass(WordCountSample.class);

job.setMapperClass(SampleMapper.class);

job.setCombinerClass(SampleReducer.class);

job.setReducerClass(SampleReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job, new Path(args[0]));

FileOutputFormat.setOutputPath(job, new Path(args[1]));

System.exit(job.waitForCompletion(false) ? 0 : 1);

}

全体構成

元となるクラス内で、内部クラスとして、Map用のクラス、Reducer用のクラスを

記述しています。

このあたりは、Hadoop Tutorialの構成を参考にしています。

Mapクラス解説

	public static class SampleMapper extends Mapper<Object, Text, Text, IntWritable> {

		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

			Tokenizer tokenizer = new Tokenizer();
			List<Token> tokens = tokenizer.tokenize(value.toString());

			for (Token token : tokens) {

				word.set(token.getPartOfSpeechLevel1());
				context.write(word, one);
			}
		}
	}

public static class SampleMapper extends Mapper<Object, Text, Text, IntWritable> {

private final static IntWritable one = new IntWritable(1);

private Text word = new Text();

public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

Tokenizer tokenizer = new Tokenizer();

List<Token> tokens = tokenizer.tokenize(value.toString());

for (Token token : tokens) {

word.set(token.getPartOfSpeechLevel1());

context.write(word, one);

}

MapクラスはMapperを拡張して作成します。

MapもReduceも共にデータをkey、valueで格納することを想定しており、

Mapperジェネリクスの型はKEYIN、VALUEIN、KEYOUT、VALUEOUTとなります。

valueはWritableインターフェースを実装した型を期待しています。

writableはhadoop内でシリアライズ、デシリアライズを行うようにするための

ものであり、データ転送に適したものである必要があるということです。

Tokenizerがkuromojiの定義でありtokenizeメソッドで解析します。

また、このmapメソッドはテキストファイルの1行ごとに呼び出されます。

kuromojiは解析すると、以下のような内容がTokenクラスに格納されます。

品詞は４つ細分類され、それぞれの読みや発音などの情報が格納されます。

Surface form	Part-of-Speech	Base form	Reading Pronunciation
私	名詞,代名詞,一般,*	私	ワタシ
の	助詞,連体化,,	の	ノ
名前	名詞,一般,,	名前	ナマエ
は	助詞,係助詞,,	は	ハ
hoge	名詞,一般,,	*	?
です	助動詞,,,*	です	デス
。	記号,句点,,	。	。

今回は細分類のうち、1番初めのものを使用しています。

その取得箇所がtoken.getPartOfSpeechLevel1()となります。

contextにwriteしている箇所は、

キーとなる品詞の数が1個存在するという形でため込んでいるということになります。

Reduce解説

	public static class SampleReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

		private IntWritable result = new IntWritable();

		public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

			int sum = 0;

			for (IntWritable val : values) {
				sum += val.get();
			}

			result.set(sum);
			context.write(key, result);

		}
	}

public static class SampleReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

private IntWritable result = new IntWritable();

public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

int sum = 0;

for (IntWritable val : values) {

sum += val.get();

}

result.set(sum);

context.write(key, result);

}

ここで行っていることは、Mapでキーごとに1個と数えたものを

集計しています。

この処理で品詞ごとの数をカウントしています。

main解説

	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();
		Job job = new Job(conf, "word count sample");
		job.setJarByClass(WordCountSample.class);
		job.setMapperClass(SampleMapper.class);
		job.setCombinerClass(SampleReducer.class);
		job.setReducerClass(SampleReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		System.exit(job.waitForCompletion(false) ? 0 : 1);
	}

public static void main(String[] args) throws Exception {

Configuration conf = new Configuration();

Job job = new Job(conf, "word count sample");

job.setJarByClass(WordCountSample.class);

job.setMapperClass(SampleMapper.class);

job.setCombinerClass(SampleReducer.class);

job.setReducerClass(SampleReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job, new Path(args[0]));

FileOutputFormat.setOutputPath(job, new Path(args[1]));

System.exit(job.waitForCompletion(false) ? 0 : 1);

}

ここでは実際の処理の準備になります。

Jobクラスを生成し、

・実行クラス

・Mapperクラス

・Combinerクラス

・Reducerクラス

・OUTのキークラス

・OUTのvalueクラス

・入力ファイル読込先

・出力ファイル出力先

を定義しています。

job.waitForCompletionにて処理完了を待つということになります。

実行する

javaアプリケーションとして実行

mainメソッドを含んでいるので、当然にjavaアプリケーションとして

実行可能です。

実行の構成にて処理引数に入力パス、出力パスと指定すればOKです。

出力パスは存在するとエラーになるので注意が必要です。

Hadoopで実行

Hadoopで実行するためには、まずEclipse側で

maven packageしてjarを生成します。

そして、以下のコマンドを実行します。

%HADOOP_HOME%\bin\hadoop jar <生成したjarパス> <フルパッケージでのメソッド名> <入力パス> <出力パス>

1	%HADOOP_HOME%\bin\hadoop jar <生成したjarパス> <フルパッケージでのメソッド名> <入力パス> <出力パス>

結果

今回の入力結果では、以下のような結果になるかと思います。

助動詞	6
助詞	18
動詞	7
名詞	17
記号	10
連体詞	2

助動詞 6

助詞 18

動詞 7

名詞 17

記号 10

連体詞 2

まとめ

今回はHadoopのMapReduceを実際に作成してみました。

Hadoopで扱えるデータの型には色々な種類があるので、

興味のある方はそれらを使って、実際に手を動かしてみると

良いかと思います。

それでは！！

にほんブログ村

人気ブログランキング

たーちゃん

20代前半までは東京で音楽をやりながら両手の指以上の業種でアルバイト生活をしていましたが、某大手プロバイダのテレアポのバイトでPCの知識の無さに愕然とし、コンピュータをもっと知りたい！と思ったことをきっかけに25歳の時にITの世界に未経験で飛び込みました。
紆余曲折を経て、現在は個人事業主としてお仕事させていただいており、10年ほどになります。
web制作から企業システム構築、ツール開発など、フロントエンドもバックエンドもサーバーもDBAも依頼があれば何でもやってきた雑食系エンジニアです。

今風にいうとフルスタックエンジニアということになるのかしら？？

→ 詳細プロフィールというか、生い立ちはこちら

→スキルシートをご覧になる場合はこちら

→お仕事のご依頼やお見積りなどお問い合わせはこちらから！