Macでnative-hadoopを使う方法

忘れないようにメモ
さて、事の発端はMacでHadoop2系のnative-hadoopを使いたくて見つけた以下の文から始まる。

The native hadoop library is supported on *nix platforms only. The library does not to work with Cygwin or the Mac OS X platform.

マジかよ... ってなこと言ってたらパッチがあるのを教えてもらう。


まず、Hadoopのソースを取ってくる。適当なところに展開しておきましょう。
また、現時点で最新のパッチ(HADOOP-9648.v2.patch)もDLしておく。

展開した先でパッチを当てる

patch -p1 < HADOOP-9648.v2.patch

native-hadoopをコンパイルするにはcmakeが必要なのでインストールしておく

brew install cmake

native-hadoopだけ欲しいので余計なビルドを実行したくないし移動

cd hadoop-common-project/hadoop-common/

nativeのコンパイルを実行する

mvn -P native compile

終了

$ ls target/native/target/usr/local/lib/
libhadoop.1.0.0.dylib* libhadoop.a            libhadoop.dylib@

後は好きなところにコピーするがよろし(make installはrule無いって怒られる)

Huahin Manager 0.1.5 and 0.2.2 released

需要の多かったCDH4 + MRv1版を含めてリリースしました。
あと、Hive REST API で某氏が欲しいと言ってた機能も追加しています。

• 0.2.2-mr1でCDH4 + MRv1の対応
• HiveのREST APIで同時HQL対応

Hiveのリリースは要するに以下のようなこと
通常は

create table foo(bar string)
select * from words
drop table foo

と複数回にわけて渡していたところを、

create table foo(bar string); insert into foo select * from words; drop table foo;

(このHQL意味ないけど)のように";"で複数同時実行を行えるようになった。
(Hive CLIと同じコードなのでバグはHive CLIが直れば直す)

Huahin Framework

Huahin Manager 0.1.4 and 0.2.1 released

HCJ2013Wまでにリリースしとうこうかなと思って。

• 0.2.1でPig REST APIに対応
• 0.1.4でHadoop 1系とCDH3のHiveとPigのREST API対応

最近ブログがリリースしか書いてない。

Huahin Framework

Huahin Manager 0.2.0 released

CDH4 と Apache 2.0.2-alpha対応でございます。
beta版から設定項目も増えてるので注意が必要です。
以下、主な出来事

• Job関連のAPIは変わってない
• YARN Application APIが追加
• Job Kill だけじゃなくて Application Kill でもいけるよ！
• それぞれのManager等が持っているREST APIにproxy接続
• HiveQLの実行(HiveServer1 or 2対応)
• 戻り値を返さない場合は、executeを使用
• 戻り値を返す場合は、executeQueryを使用
• 戻り値はストリームで返される
• 実行時間がアレだとアレなのでタイムアウトには気をつけてね
• Job の キュー登録にHiveが追加
• キューでの実行なので出力先はテーブル、HDFSにする必要あり

たぶん、次はPigが入るかなと、0.1系もHiveなコードは入ってるんだけどpushをしてないのでそのうちにあれするかなと。

Huahin Framework

Huahin Manager 0.2.0-beta released

さて、皆さんお待ちかね Manager の CDH4 対応です。
存分にステマするが良いと思います。

今回、beta としてるのはCDH4のクラスタが無いのでクラスタでのテストが出来てないってこと...
なのでbetaとして出してクラスタ持ってる人にテストしてもらおうかなと思って。。

あとMRv1には対応してるけど、YARN対応はまだなのとApplication Masterとかに対応してないのでbetaが取れる時に入る感じで。まだCDH4しか対応してないのでApache版もその時に...

Huahin Framework

Huahin EManager 0.1.0 released

EManager、Managerと何が違うのかっていうと、Amazon Elastic MapReduce(EMR)に特化したManagerです。もちろん、EManagerのEはEMRのEです。

本当はManagerの中に入れちゃおうと思ったんだけど、いろいろ無理が出る感じがしたので、分けたしだいです。まぁ、内部での使用もEManagerとManagerがあって、インターフェースで切り替えて、オンプレHadoopとEMRを使い分けてる感じなので、同じっちゃ同じなんですけど。

EManagerができることは、基本Managerと同じで以下のとおり。

• Job Flowの一覧取得
• Job Flowの詳細取得
• StepのKill
• キューの一覧取得
• キューの詳細取得
• キューの登録

まず、Managerと違う点。

Jobの一覧等がJob FlowとStepという感じでEMRのものになってます。キューはManagerと同じだけど、Jobの実行できるものがHive, Pig, Streaming, Custom JarとEMRで使えるものは基本的に使えます。

EManager独自の機能。

まず、Stepのkill。現状、EMRはJobのkillを行うにはEMR自体をターミネートするか、マスターインスタンスにログインしてkillするしか方法がありません。EManagerはManagerと組み合わせることでこの機能を実現しています。

クラスタ数の指定。複数のHadoopクラスタ(EMR)を起動し制御することができます(複数起動する場合は、インスタンスの上限値に注意する必要あり)。

コスト管理。EC2は時間課金なので1時間使用しようが5分使用しようが料金は同じです。Jobが終了したらEMRをターミネートするのではなく、起動したままで次のキューが登録されたら、起動しているEMRにJobを割り振ります。1時間未使用のままであれば自動的にEMRがターミネートされます。

Job(step)の実行。EMRはS3に保存してあるスクリプトなりを指定して実行しますが、EManagerのREST API(APIしかないんだけど)のメソッドでPOSTを指定した場合にファイルをS3に送信することができるのでオンデマンドで実行する時に楽です。オプションの指定で、実行が完了したスクリプトは削除もできます。

などなど。

一応これで、内部のツール系ものは出し切った感。

Huahin EManager

Processing small data by Hadoop

ビッグデータ()が流行ってる昨今ですが、あえてスモールデータを処理する話題でも。

ビッグデータってさぞかしデカいデータでしょうって話がよく聞かれるわけですよ。
（最近はあんまり聞かないのでもうみなさんあきてきましたか？）
◯TB、◯PB、えぇ、確かにデカいです。こんなん処理するのにどのぐらい掛かるんですか？って話もあります。

んじゃ、逆を言えばこれらからみると数百MBぐらいのデータって小さいですね。
中身も100万件ほどしかなかったり。
単純に集計やる程度だったらHadoopとか必要ないですね。ってか、誰がそんなめんどかいことやるのかと。

じゃあ、ちょっと考えてみる。
100万件のデータがあります。全部1件ずつそれぞれ処理をしなければいけません。1件処理するのに大体0.1秒掛かったとします。

1,000,000(件) * 0.1(秒) / 60 / 60 = 27.77...時間

これ、スモールデータですか？1日じゃ処理しきれませんよ...
これをマルチスレッドで並行してやったとしても

約28時間 / 8(4コア X 2の想定) = 3.5時間

それでも3.5時間掛かるわけです。

まぁ、こんな話をしようと思ってたわけじゃないのでこの辺りでやめておいて本題。


上記のような処理をHadoopでやらせようと思った場合、単純にMapがネックになる。例えば、上記のものが300MBだったとする。HDFSのデフォルトのブロックサイズは64MB。Mapはブロック単位で処理されるので

300 / 64 = 5(約4.6)

ってことは、Mapは5個しか起動されないわけです。20ノードとか並べてもまったく意味がないわけですね。。。これを回避する方法としてInput部分を自分でアレするってのがありますが、正直そんなのめんどくせぇよってなるわけです。ってことでお手軽な方法として。

FileInputFormat.setMinInputSplitSize(job, 1048576);
FileInputFormat.setMaxInputSplitSize(job, 1048576);

とかやって処理するサイズをちっちゃくしてあげましょう。


次に数GB程度のファイルを成形する処理を考えてみましょう。
先に書いちゃうとローカルモードのHadoop速いです。速いです...
Hadoopのイニシャルコストって結構掛かるのですぐ起動するし速いです。

んでは、やり方。

普通(分散)のHadoop同様Mapがブロック単位で処理されます。ここでネックになってくるのがMapが1つずつ起動するってことです。ようするに一つ目のMapが終了すると次のMapが起動する。タスクの起動コストが高くとても効率が悪いです。。。

上記とは逆でsplitサイズをデカい値にしちゃってMapを一つしか起動しないようにしちゃいます。

FileInputFormat.setMinInputSplitSize(job, 2147483648);
FileInputFormat.setMaxInputSplitSize(job, 2147483648);

次に、これだとシングルスレッドになってしまうのでマルチスレッド化しちゃいましょう。

job.setMapperClass(MultithreadedMapper.class);
MultithreadedMapper.setMapperClass(job, Mapper.class);
MultithreadedMapper.setNumberOfThreads(job, 4);

MultithreadedMapperはMapperをマルチスレッド化してくれるので自分でマルチスレッドの処理を書かなくてすむのでとても楽ですね。
ただ、障害が起きると即死亡なのでそこは気をつけましょう。

ローカルで処理しきれないぐらいでかくなってきたらクラスタにしちゃえばいいので一粒で二度美味しい的な感じがあります(まさにビッグデータ、スモールスタート！)。

本当はここだけをメモ代わりに書いとこうと思ったんだけど、余計なことまで書いちゃったのはもういいや。

ということで、上記のファイルの処理だけやるヤツを公開しました。

Huahin Tools

まだ、ファイルを成形(デフォルトではデフォルトのApacheログを成形)する機能だけしかないですが、成形してからDBなんかにぶち込む感じで。
現在はローカルモードとAmazon Elastic MapReduceで動くようになってます。

注意としては、他のものとリリースをあわせようと思ってるので、まだ正式版ではないってこと。

Huahin Framework(wrapping Hadoop MapReduce) released

内部で使用しているフレームワークをOSS化することにしました。

Huahin Framwork (http://huahinframework.org/)

Huahin Manager (http://huahinframework.org/huahin-manager/)に関しては某所で求められてて先行でリリースしていたのですが、こちらはMapReduceをラップしたフレームワーク本体です。

Pig、Hiveじゃどうしてもできないことがあったためネイティブで書くしかなかったのですが、正直、Writableとかソートとか書くのにはウンザリしていたためこれが出来た感じです。

基本は、MapReduceをラップしてるだけなので足りないものは自分で実装できるようになってます。Key/Valueがレコードという概念になっていて、レコードに対してグルーピングするのかソートにするのかそれともただの値なのかといった感じで指定します。

ジョインに関して内部仕様からの切り出しに苦労してるため現在はサポートされてませんが(自分で実装するのであればできる)、次のリリースでサポートする予定です。

Huahinとは何かってことですが、タイのHua Hinから来ています。
ウチでつけるコード名はワインの産地から決めるって感じなのですが、ワイン産地、タイ＝象ということでコード名がそのままフレームワーク名になった超適当な名前付けです。





OSS化って、内部仕様どっぷりだと結構てこずるますね。
去年のHadoop Conference Japanで話してから随分時間が掛かった。。

MRUnit 0.9.0 released (mavenでの取得方法)

ついでなんでメモがわりに書いておく。

MRUnitの0.8.1からHadoop 0.23を対応したことによりMavenでの
取得方法が変更になっている。

0.8.0までだと

<dependency>
  <groupid>org.apache.mrunit</groupid>
  <artifactid>mrunit</artifactid>
  <version>0.8.0-incubating</version>
  <scope>test</scope>
</dependency>

のような感じでよかったのだけど、0.8.1からHadoopのバージョンを指定してあげないといけない。
下記のような感じでclassifierにバージョンを指定してあげる。

<dependency>
  <groupId>org.apache.mrunit</groupId>
  <artifactId>mrunit</artifactId>
  <version>0.9.0-incubating</version>
  <classifier>hadoop1</classifier>
  <scope>test</scope>
</dependency>

ちなみに、hadoop1は1.0.2でhadoop2を指定すると0.23.1が指定されることになる。

Hadoop Hacks now on sale

ステマ的には遅いが一応書いとく。

が4/25に出ました。
まぁ、いろいろありました。

右の翻訳と一時期かぶってたりとか(結構キツかった)。
某氏が起業してベイエリアに行っちゃうとか。
代わりにuしんさんひっぱってきたりとか。

無事出てよかったですね。

HH2をOSSとしてリリースします

HadoopにかわるHH2の開発を進めてきましたがOSSとしてリリースすることにしました。

HH2はHadoopのJobTracker、TaskTrackerを置き換えるもので効率よくJobを実行することができます。まず、JobTrackerにかわるHumanTrackerがTaskTrackerにかわるHumanTaskerにタスクの割り当てを行います。

タスクの割り当てに使うのは特に難しいことはありません。メールによる通知となります。

この時、通常のHadoop同様にHumanTaskerがなんらのか障害により通達エラーが発生する場合があります。

この場合もHadoop同様に別Humanにタスクを割り当てます。ただし注意しなければいけないのは別Humanにタスクを割り当てる場合は、ボーナスの支給が必須となります。

最後に、やはりHumanTrackerはHadoopのJobTracker同様にSPOFになりえるということです。HumanTrackerの障害は様々ですがいずれこの問題は解消される予定です。

ここまで見てきたとおり通常のHadoopよりも効率的にJobが実行できることがおわかりいただけたでしょう。なお、HH2のOSSリリース予定は2045年5月を目標に着々と準備を進めているところです。

リリースが完了するまでしばらくお待ちいただけますと幸いです。

HH2開発チーム一同

Hadoop Conference Japan 2011 Fall

内容は書く気力が現在無いためアップロードだけで。。。人材は本当に募集しておりますです。
Hadoop Conference Japan 2011 Fall

View more presentations from Ryu Kobayashi

第3回Hadoopソースコードリーディング Hadoop MR + Cassandra のスライド #hadoopreading

昨日使ったスライドをアップしました。

若干修正しました。
失念していたissueのところの追加とデモで使用した、
GUIの公開先を追加しました。

Hadoopソースコードリーディング第3回 Hadopo MR + Cassandra

View more presentations from Ryu Kobayashi.

Hadoop第2回ソースコードリーディングのスライド

昨日のスライドを上げました。

Hadoopソースコードリーディング MapReduce障害時のフロー

View more presentations from Ryu Kobayashi.

修正したい気もするがメンドクサイのでもういいｗ
ってか、ソースも張らないとなんのこっちゃって感じもしますが。
手元のソースご覧くださいってことで。

MapReduceのテスト

MapReduceのテストはHadopp本にはMockitoを使う良いあるよと書いております。
でも、やはり0.19.xが前提なので情報が古い。
ってことで、0.20.xで以下のような感じで書いてみた。

package example.test;

import java.util.Arrays;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import static org.mockito.Mockito.*;

import junit.framework.TestCase;

public class CountTest extends TestCase {
    @SuppressWarnings("unchecked")
    public void testMapLongWritableTextContext() {
        // Mapのテスト
        Map map = new Map();
        Mapper.Context map_context = mock(Mapper.Context.class);

        Text value = null;
        try {
            value = new Text("test1");
            map.map(null, value, map_context);
            verify(map_context, times(1)).write(new Text("test1"), new LongWritable(1));

            value = new Text("test2");
            map.map(null, value, map_context);
            verify(map_context, times(0)).write(new Text("test1"), new LongWritable(0));
        } catch (Exception e) {
            fail(e.getMessage());
        }


        // Reduceのテスト
        Reduce reduce = new Reduce();
        Reducer.Context red_context = mock(Reducer.Context.class);

        Text key = null;
        Iterable<longwritable> values;
        try {
            key = new Text("test1");
            values = Arrays.asList(new LongWritable(2));
            reduce.reduce(key, values, red_context);
            verify(red_context, times(1)).write(new Text("test1"), new LongWritable(2));

            key = new Text("test2");
            values = Arrays.asList(new LongWritable(2), new LongWritable(2));
            reduce.reduce(key, values, red_context);
            verify(red_context, times(1)).write(new Text("test2"), new LongWritable(4));
        } catch (Exception e) {
            fail(e.getMessage());
        }
    }
}

HDFSへの書き込み/読み込み方法

下記のような感じで。

Configuration conf = new Configuration();
// ユーザを指定する場合
conf.set(UnixUserGroupInformation.UGI_PROPERTY_NAME, "hoge,supergroup");
FileSystem fs = FileSystem.get(new URI("hdfs://hoge:9000/"), conf);

// ディレクトリの作成など
try {
  fs.getFileStatus(new Path("hoge"));
} catch (FileNotFoundException e) {
  fs.mkdirs(new Path("hoge"));
}

// ファイルへの出力
// ex. 最後のフラグは上書き
FSDataOutputStream dos = fs.create(new Path("hoge/hoge.txt"), true);
dos.writeChars("テスト\n");
dos.close();

みたいに。

Hadoop MapReduce 0.19.* -> 0.20.*

Hadoopネタが続いてるが備忘録なので

MapReduce の0.19.*から0.20.*でかなり仕様が変わっている。

情報も少ないし、本家のチュートリアルも古いしでまったく使い物にならない。
ってことで、基本部分の変更箇所。

org.apache.hadoop.mapredで
使用していたものが
org.apache.hadoop.mapreduce

を使用するように変わっている。

public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
  public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter)       
    throws IOException {
  }
}
public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
  public void reduce(Text key, Iterator<Intwritable> values, OutputCollector>Text, IntWritable< output, Reporter reporter)
    throws IOException {
  }
}

みたいにMapもReduceもインターフェースでやってたいところが

public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
  protected void map(LongWritable key, Text value, Context context)
    throws java.io.IOException ,InterruptedException {
  }
}
public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
  protected void reduce(Text key, Iterable<intwritable> values, Context context)
    throws java.io.IOException ,InterruptedException {
  }
}

みたいに抽象クラスを使用するようになっている。
# 基本、上記の古いほうはDeprecatedになってる。

で、Jobもそのまま使用するようになっているのでだいたい、こんな感じに変更。

package example.hadoop;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;

public class UserCount extends Configured implements Tool {
  public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    protected void map(LongWritable key, Text value, Context context)
      throws java.io.IOException ,InterruptedException {
      String[] s = value.toString().split(",");
      word.set(s[0]);
      context.write(word, one);
    }
  }

  public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable result = new IntWritable();

    protected void reduce(Text key, Iterable<intwritable> values, Context context)
      throws java.io.IOException ,InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }

  @Override
  public int run(String[] args) throws Exception {
    Job job = new Job(getConf(), "user count");
    job.setJarByClass(UserCount.class);

    job.setMapperClass(Map.class);
    job.setCombinerClass(Reduce.class);
    job.setReducerClass(Reduce.class);

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

    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);

    return job.waitForCompletion(true) ? 0 : -1;
  }

  public static void main(String[] args) {
    status = ToolRunner.run(new UserCount(), args);
    System.exit(status);
  }
}

以上、備忘録。

Task process exit with nonzero status of 1.が出る件

エラーの出るサーバをスレーブからはずし起動するとこのエラーが出ない。
ということでOSの問題もあるなこりゃ。


どっかでインストールし直したほうが良いと。

HDFSでスレーブからマスタに接続できない件

たぶん解決。

なんのことはない、IPv6の接続がやっぱりネックになっていた。
hadoop-env.shのHADOOP_OPTSを

export HADOOP_OPTS="-server -Djava.net.preferIPv4Stack=true"

みたいにIPv4でに全てのマスタ/スレーブ設定すればおｋ。


# 追記
やっと起動したと思ってMapReduceを何回か実行すると

java.io.IOException: Task process exit with nonzero status of 1.
  at org.apache.hadoop.mapred.TaskRunner.run(TaskRunner.java:418)

が出る。ヽ(｀Д´)ノ
コレか？

http://www.koopman.me/2009/04/hadoop-0183-could-not-create-the-java-virtual-machine/

Hadoopの

HDFSでスレーブからマスタに接続できない。

core-site.xmlの

  
    fs.default.name
    hdfs://localhost:9000
  

のvalueを

  hdfs://hoge:9000
  hdfs://hoge.example.jp:9000
  hdfs://0.0.0.0:9000

等、いろいろ試したが全てダメ。

ERROR org.apache.hadoop.hdfs.server.datanode.DataNode: java.io.IOException: Call to hoge/192.168.0.1:9000 failed on loc
al exception: java.net.SocketException: Network is unreachable

となりまふ。

Debian/Ubuntu間でやってるけどIPv6のせいかと思い無効化してみるけどやはりダメ。
マスタサーバ内にスレーブを立ち上げは繋がる。

何が原因なんだろうかと投稿し誰かの助け舟を待ってるみる。