No Such Blog or Diary

今回の論文を書くための実験で，複数のHadoopのクラスタセットひとつひとつに対応するターミナルを開き，それぞれで似たようなコマンドを実行して実験結果を取るという作業をした．コマンドをカットアンドペーストで全部のターミナルに張り付けて実行するとかバカすぎる．なんとかしたい．

という背景で，複数のターミナル（シェル）に対して一箇所から機械的にコマンドを打ち込みたい要求に駆られた．pssh とかだとインタラクティブに出来ないし，そもそも出力のためのターミナルは複数のウインドウにわかれていて欲しい．やりたいのは入力の一本化だけ．

単純に考えると，fifo を使って入力ターミナルからのコマンドを出力ターミナルの標準入力に送ればいい気がする．なので，とりあえず，出力用ターミナルで

cat pipe | bash

としておいて，入力用ターミナルでコマンドを

echo 'cd hoge' > pipe
echo 'ls' > pipe

とかやって送り付けようとした．

が，失敗．何故かコマンドをひとつ発行しただけで出力ターミナルの bash が終わってしまう．これでは複数のコマンドをインタラクティブに発行できず，psshと変わらない．これは望むものではないので原因を探って手法を改善せねばらない．

上の単純な仕組みが失敗した原因は次の通り：最初に送られたコマンドを出力側の bash が読み込んだとき，pipe が EOF に達してしまって "cat pipe | bash" のコマンドが終了してしまう．pipe には後から内容が追加されるけど，とりあえず最初のコマンドの後に一度はファイルの終端まで達するので，シェルのパイプが終了してしまう，と．

さてどうしたものか．

適当に検索してみたら同じようなことを考える人はいたようで，解決策が見つかった．"cat pipe | bash" の代わりに次を使えばいい．

tail -f pipe | bash

この tail の -f オプションは，EOFに達しても tail を止めずEOF後に追加された内容も後ろに吐く，という指定．手前のコマンドの出力のプログレスを表示するとかの目的に使われるオプションらしい．これをやっておけばコマンドを間隔を置いて複数送り付けることが出来る．

ということで，前準備＠入力ターミナル：

ps="1 2 4"
for p in $ps; do mkfifo pipe-$p; done

前準備＠出力ターミナル達：

tail -f pipe-1 | bash

tail -f pipe-2 | bash

tail -f pipe-4 | bash

あとは，適宜全部のターミナルに対して for 文使って少しずつ違うコマンド送ったり，一斉にコマンド発行したり，特定のターミナルにだけ cat 使ってインタラクティブに仕事したり，tee と合わせて特定のサブセットにだけコマンドを送ったり，等など．それなりに使い勝手が良い気がする．

for p in $ps; do echo "procs=$p" > pipe-$p; done
for p in $ps; do echo 'this is pipe-$procs' > pipe-$p; done
for p in $ps; do echo "./runExp.sh" > pipe-$p; done
...
cat | while read line; do echo $line; done > pipe-4
...
cat | while read line; do echo $line; done | tee pipe-2 | tee pipe-4
...
for p in $ps; do echo "cat result-$p.txt" > pipe-$p; done

cat の while read line は１行単位でコマンドを送るため．ここらへんは関数を定義しておいたほうが頭よさげ．

あとは最初のセットアップのところを for 文でいけると楽．ということで，出力ターミナルを一斉に作ってパイプをつなぐコマンド：

for p in $ps; do (echo "(echo 'echo this is pipe-$p'; tail -f pipe-$p) | bash" > pipe-$p &) ;( gnome-terminal -x bash pipe-$p &); done

これで満足．

ひとつのファイルは HDFS 上に物理的にチャンクに区切られて置かれる．FileSplit はひとつのファイルを論理的に区切ったもの．ひとつの FileSplit に対して Map のジョブひとつが対応して実行される．細かくは，ひとつの FileSplit がひとつの RecordReader を生成し，その RecordReader がその Map ジョブへと Key/Valueペアのリストを提供する．このとき，その RecordReader はその FileSplit で指定された論理的な領域を越えて（HDFS上にあるだろう）元のファイルを読み込むことが出来る．

一般に，HDFSのチャンクはFileSplitではないし，FileSplitとMapジョブに提供される仕事とは完全に一致はしない．

実際，デフォルトのTextInputFormatが作るRecordReaderは，与えられたFileSplitの最終行を完成させるため，次のFileSplitの境域へと境界を越えて最後の行を読みに行く．例えば，改行なしの大きなファイルを入力とすれば，最初のFileSplitから作られたRecoardReaderは改行を求めてファイル全部を読んでしまう．結果として，先頭のMapジョブはファイル全体を入力として受け取る．

大体にしてインストールというかセッティングが面倒なので手順のメモ．

1. ダウンロードして展開．クラスタの場合，本体の置き場所は NFS 上にしとくとインストールが楽（ログのディレクトリに注意）．さもなければ全マシンの同じパスに置いておくべし（パスをマシンごとに変えていいのかどうか分からん）．

   wget http://ftp.kddilabs.jp/infosystems/apache//hadoop/core/hadoop-0.21.0/hadoop-0.21.0.tar.gz
   tar xfvz hadoop-0.21.0.tar.gz
   cd hadoop-0.21.0

2. 環境ファイル conf/hadoop-env.sh の編集：JAVA_HOME と HADOOP_HEAPSIZE と HADOOP_LOG_DIR．

   ログの出力先はデフォルトで hadoop の置いてあるディレクトリ(HADOOP_HOME)の下の logs ディレクトリなので，これが NFS 上だったりすると酷いことになる．なので，ローカル上に取るように指定を入れておく．

   export JAVA_HOME=/usr/lib/jvm/java-6-sun/
   export HADOOP_HEAPSIZE=2048
   export HADOOP_LOG_DIR=/tmp/user/node000/logs
   

3. 設定ファイル conf/slaves：計算に使うノード名を列挙．

   node000
   node001
   node002
   node003
   ...
   

4. 設定ファイル conf/core-site.xml の編集．

   namenode として使う計算機の名前（URI）を書いておく．マルチコア計算機１台なら localhost で十分．クラスタの場合には localhost ではまずいので，ちゃんとした計算機名前を書いておく．

   <configuration>
        <property>
            <name>fs.default.name</name>
            <value>hdfs://node000:9000/</value>
        </property>
   </configuration>
   

5. 設定ファイル conf/hdfs-site.xml の編集．

   dfs.name.dir と dfs.data.dir に /tmp とかローカルなディスクのパスを書いておく（設定書かなくてもデフォルトで /tmp に適当なディレクトリを作って使ってくれるけど）．NFS上のパスを指定するのは何考えているか分からない．あとは dfs.blocksize （HDFS上のチャンクのサイズ）を適当に変えてもいいかも知れない．Mapper への一仕事の最大値はこのサイズ．

   <configuration>
        <property>
            <name>dfs.name.dir</name>
            <value>/tmp/user/node000/name-dir/</value>
        </property>
        <property>
            <name>dfs.data.dir</name>
            <value>/tmp/user/node000/data-dir/</value>
        </property>
        <property>
            <name>dfs.blocksize</name>
            <value>4194304</value>
        </property>
   </configuration>
   

6. 設定ファイル conf/mapred-site.xml の編集．

   とりあえず，mapred.job.tracker に JobTracker のいる計算機の名前を書いておく．マルチコア１台なら localhost で十分．クラスタなら計算機名をちゃんとかく．あと，計算が軽い時には JVM を使い回さないと遅くて困るので mapred.job.reuse.jvm.num.tasks に無限回の使い回しを意味する -1 を入れておく．ノード一つ当たりのジョブの数は mapred.tasktracker.map.tasks.maximum と mapred.tasktracker.reduce.tasks.maximum で指定しておく．マルチコアならコア数以上のジョブが同時にあって構わないので，コア数以上の数字を書いておく（2コアマシンしか無いのに200とか書くとプロセスが多すぎて死ぬ）．また，Hadoop ではひとつのファイルを複数の FileSplit に分割し，その FileSplit １つに対して Mapper が１つ呼ばれて動く（間に RecordReader が挟まって「FileSplit→KVペアの集合」という変換が入るけど）．その FileSplit のサイズ指定が mpreduce.input.fileinputformat.split.maxsize でできる．実際には，これで指定したサイズと HDFS のチャンクサイズとの小さいほうが実際の FileSplit のサイズになる（全タスク数の指定を入れたときには，さらにそこから導かれるサイズとの小さいほうかね）．

   <configuration>
        <property>
            <name>mapred.job.reuse.jvm.num.tasks</name>
            <value>-1</value>
        </property>
        <property>
            <name>mapred.job.tracker</name>
            <value>node000:9001</value>
        </property>
        <property>
            <name>mapred.tasktracker.map.tasks.maximum</name>
            <value>2</value>
        </property>
        <property>
            <name>mapred.tasktracker.reduce.tasks.maximum</name>
            <value>2</value>
        </property>
        <property>
            <name>mapreduce.input.fileinputformat.split.maxsize</name>
            <value>4194304</value>
        </property>
   </configuration>
   

7. テスト．

   namenode を初期化して，他のノードを起動して，最初にディレクトリ作って，そこにファイルを転送して，サンプル動かして，出力確認して，邪魔な出力消して，そしてノード停止．

   bin/hadoop namenode -format
   bin/start-all.sh
   bin/hadoop fs -mkdir input
   bin/hadoop fs -put conf/* input
   bin/hadoop jar hadoop-mapred-examples-0.21.0.jar grep input output 'dfs[a-z.]+'
   bin/hadoop fs -cat output/*
   bin/hadoop fs -rmr output
   bin/stop-all.sh
   

これで言う事聞くようになった．

その他無茶な設定 in conf/mapred-site.xml：なるべくディスクつかなわいように無理をする．

     <property>
         <name>io.sort.record.percent</name>
         <value>1.0</value>
     </property>
     <property>
         <name>io.sort.spill.percent</name>
         <value>1.0</value>
     </property>
     <property>
         <name>io.sort.mb</name>
         <value>4</value>
     </property>
     <property>
         <name>mapred.inmem.merge.threshold</name>
         <value>2048</value>
     </property>
     <property>
         <name>mapred.job.reduce.input.buffer.percent</name>
         <value>1.0</value>
     </property>
     <property>
         <name>mapred.job.shuffle.input.buffer.percent</name>
         <value>1.0</value>
     </property>
     <property>
         <name>mapred.job.shuffle.merge.percent</name>
         <value>1.0</value>
     </property>

安田講堂前に立派な雪像が出来ていた．

誰が作ったか分からないけど，雪の台座（高さ１ｍ位）を作っているところが秀逸．雪像自体も気合が入っている．携帯のカメラなのが残念…

ラインごとに Mapper に渡さなくていい，まとめて寄越せ　→　自前で FileSplit 全体を Mapper に与える RecordReader を実装．中身はすっからかんだけどね．

bin/hadoop namenode -format して bin/start-all.sh して bin/hadoop fs -put hoge.dat . とかすると文句を言われる　→　bin/hadoop fs -mkdir tmp してから bin/hadoop fs -put hoge.dat . すると文句を言われない．からの状態からディレクトリを何でもいいからひとつ作ってやると，ホームディレクトリ（.）にファイルを置ける．というか，mkdir でディレクトリ作ったときにホームディレクトリもついでに作られるからなのか？　とにかくディレクトリを先に作っておかないとファイルを置けないのかね？

mapreduce の FileSplit のサイズは mapred-site.xml でmapreduce.input.fileinputformat.split.maxsize を指定．

bin/hadoop --config conf-hoge ... とかで設定ファイルを切り替えられる．start から stop まで一貫して同じ設定を使わないと色々とおかしくなってくれるけど．

曽我の梅林へ梅を見に行った．１０時過ぎには大学にいく必要があったので，日の出前に梅林に到着．路面が凍結しているので危ない．特に止まった時が一番危ない．気を受けないと付いた足が滑って立ちゴケする．コケなかったけど．走行距離２００キロ．

んで，別所梅林．ここらの梅林全体をまとめて曽我梅林と呼んで，別所梅林はその中のひとつってことでいいのかな．よく解らんけど蘇我の梅林に行こうと言っていつも来ていたのはここだったはず．

雪の降った翌日の早朝なので，寒いシリーズ：赤富士，雪をかぶった箱根の山，朝日．

蘇我の梅林は好き勝手に入り込める点が良い．好きなように梅が見られるし，地面の草も良い感じ……　寒すぎて凍ってたれど．昼間ならきれいな緑の地面と青い空と白い梅とか撮れるはず．でも地面が凍っているので白い梅と早朝の青空．

そして帰りの東名高速で事故のために渋滞発生．おかげで大学に行くのが微妙に遅れてしまった．迷惑だから自爆でも事故るな．