Hadoopで処理を実装してみる──Hadoop Streamingでの処理、script-runner.jarの使い方：きょうから試せる Hadoop“スモールスタート”ガイド（8）（4/4 ページ）

[佐々木達也，著]

ユニークユーザ数の算出

　ではもうひとつ別の例としてユニークユーザ数の算出を行ってみましょう。入力データはこのようなスクリプトで3000万件の仮想ユーザIDを準備します。

create_user_id.rb

File.open('input_user_id', 'w') {|f|
  30000000.times do
    user_id = rand(100000)
    f.puts user_id
  end
}

　これでinput_user_idというファイルが生成されました。

input_user_id

80822
38755
41005
50440
67176
58612
59510
13525
40692
24805
...

　これを入力データとして、ユニークユーザ数を算出するMapper、Reducerがこちらになります。Mapperはほとんど何もしてないように見えますが、ここが実は重要な処理となっています。Mapperで"key<タブ>value"という形式で出力すると必ず同じReducerに渡されます。さらにReducer内では自動的に並び替えが行われます。つまり、同じユーザIDは必ず同じReducdrに渡されて並び替えられるため、処理する際には必ず連続して登場してくるというわけです。

uu_mapper.rb

ARGF.each do |user_id|
  user_id.chomp!
  puts "#{user_id}\t1"
end

uu_reducer.rb

pre_user_id = nil
 
ARGF.each do |log|
  log.chomp!
  user_id = log.split(/\t/)[0]
  if pre_user_id
    if pre_user_id == user_id
      # 何もしない
    else
      puts "#{user_id}\t1"
      pre_user_id = user_id
    end
  else
    puts "#{user_id}\t1"
    pre_user_id = user_id
  end
end

　ではこれらのファイルをS3に配置し、実際にこの処理をEMRクラスタ上で実行してみましょう。まずはジョブフローを起動します。

　このジョブフローに対してユニークユーザ数を計算するステップを追加します。

　正常に動作すればこのようなファイルが出力されているはずです。あとはこのファイルの行数を数えればそれがユニークユーザ数です。

part-00000

10000    1
10001    1
10002    1
10003    1
10004    1
10005    1
...

Mapperの数とReducerの数

　ところで、ここまではm1.smallを使ってEMRクラスタを構築し、処理を行ってきました。3.6.2節でも少し触れましたが、EMRはインスタンスタイプによって最適なデフォルト値を設定されています。例えばMapperの数とReducerの数についてはインスタンスタイプごとにこのように設定されています。

Mapperの数とReducerの数

	インスタンスタイプ	Mapperの数	Reducerの数
	m1.small	2	1
	m1.large	3	1
	m1.xlarge	8	3
	c1.medium	2	1
	c1.xlarge	7	2
	m2.xlarge	3	1
	m2.2xlarge	6	2
	m2.4xlarge	14	4
	cc1.4xlarge	12	3
	cc2.8xlarge	24	6

　MapperやReducerの数はmapred.map.tasksやmapred.reduce.tasksといった値で指定することで可能ですが、これらの数を増やすとよりたくさんのリソースが必要となります。EMRではインスタンスタイプによって性能が決まっているので、実質的にMapperやReducerの数を増やすのは難しいでしょう。

　そのため、今回のようにm1.smallを使っていてマスターインスタンスタイプ1台、コアインスタンスタイプ1台のような構成だとReducerが1台しか起動しません。

1台のみで処理した場合

　ただ、これではHadoopを利用する恩恵をほとんど受けられないのでタスクインスタンスタイプを2台追加してReducerが3つ起動する状態にしてみましょう。

　なお、Reducerの起動数はリサイズ時ではなくジョブフロー起動時に決定されてしまうため、ここでは明示的にmapred.reduce.tasksを3と指定する必要があります。

　これでReducerが3台起動して分散処理が行われる状態となりました。まったく同じ処理をしていますが、1台のときは17分近く掛かっていた処理が3台だと6分ちょっとと、ほぼ線形に処理性能がスケールしていることがわかります。

3台で処理した場合

　今回は入力データのサイズが小さいので1台でもあまり時間が掛かっていないですが、Reducerの数はHadoopを扱う上で非常に重要なのでいくつ動いているかは常に注意してください。

Hadoopファーストガイド

佐々木達也著
秀和システム 2400円（税別）
「ビッグデータ（＝従来のコンピュータシステムでは、しまったり、探し出したり、調査したり、人間にわかりやすく加工したりが難しい、とても大きなデータ）なんて自分には関係ない」そう思っているエンジニアに贈る「Hadoop」の体験型入門書です。数億のユーザを抱えるSNSの利用データのように「雲をつかむような話」ではなく、Webサイトエンジニアや普通のプログラマが親しみを感じられるような普通のデータと、Amazon EMRという手軽な実験環境を使い、実際にHadoopを動かしてみます。手軽に短い時間で分散処理のメリットを体験することで、手元の業務に眠っているかもしれないビッグデータの活用に可能性を見出せるかも知れません。