#!/usr/bin/perl

use strict;

while (<STDIN>) {
        chomp $_;
        print ( $_ . "\n" );
}

#!/usr/bin/perl

use strict;

my ($key0,$value) = ();
my ($key,$id) = ();
my $co = 0;
my $corr = 0;
my %co_count = ();
my %key_count = ();
my $key_current = "";
my @list = ();

while (<STDIN>) {
        chomp $_;
      ＃入力したデータをkey0とvalueのペアに分割する。分割するデリミタはMap関数の出力形式に合わせる。
      ＃ここでは前出のMapperのkey及びvalueがタブ区切りで出力したので、デリミタをタブにする。  
        ($key0,$value) = split(/\t/,$_);
      ＃key0はアイテムペアなので、$key及び$idのアイテムに分割する。分割するデリミタは第二段階のReducerの出力形式に合わせる。	
        ($key,$id) = split(/:/,$key0);

      #key値が変わった＝以前のkeyに関するデータの読み込みが終了のサインなので、以前のkeyである$key_currentに対する処理を開始する。
        if ( $key ne $key_current )  {
                if ($key_current ne "" && ( keys (%co_count) > 0 ) ) {
                        foreach $id ( sort keys %co_count ) {
                 #アイテムペアをアクセスしたユーザ数をダブルカウントしているので、二で割る。    
                                $co = $co_count{$id}/2;
                                $corr = $co/($key_count{$key_current}*$key_count{$id});
                #基準とするアイテムのID、ペアとなったアイテムID2、基準とするアイテムをアクセスしたユーザ数、両アイテムをアクセスしたユーザ数、ペアとなったアイテムをアクセスしたユーザ数、両アイテムの相関係数の順に出力 	  
                                print ($key_current . "\t". $id . "\t" . $key_count{$key_current} . "\t" . $co . "\t" . $key_count{$id} . "\t" . $corr . "\n");
                        }
                }
            #比較対象となるkeyを変更する
               $key_current = $key;
               $corr = %key_count = %co_count = @list = ();
        }
        if ( $value =~ /(\d+)>(\d+)>(\d+)/ ) {
                if ( $1 > 0 ) { $key_count{$key} = $1; }
                $co_count{$id} += $2;
                if ( $3 > 0 ) { $key_count{$id} = $3; }
        }
}
#標準入力を読み終えた段階で、まだ出力していないkeyに対し、集約を行い、その結果を出力する。
if ($key_current ne "" && ( keys (%co_count) > 0 ) ) {
        foreach $id ( sort keys %co_count ) {
                $co = $co_count{$id}/2;
                $corr = $co/($key_count{$key_current}*$key_count{$id});
                print ($key_current . "\t". $id . "\t" . $key_count{$key_current} . "\t" .
 $co . "\t" . $key_count{$id} . "\t" . $corr . "\n");
        }
}

hadoop dfs -rmr gihyo/;
hadoop jar /opt/hadoop/contrib/streaming/hadoop-*-streaming.jar - D mapred.task.timeo
ut=300000 -input amazon -output gihyo/ -mapper './mapper_for_gihyo_1.pl' -reducer
 './reducer_for_gihyo_1.pl'  -file './mapper_for_gihyo_1.pl' -file './reducer_for_gihy
o_1.pl' -jobconf mapred.reduce.tasks=12;
hadoop dfs -rmr gihyo2/;
hadoop jar /opt/hadoop/contrib/streaming/hadoop-*-streaming.jar - D mapred.task.timeo
ut=300000 -input gihyo/ -output gihyo2/ -mapper './mapper_for_gihyo_2.pl'
 -reducer './reducer_for_gihyo_2.pl' -file './mapper_for_gihyo_2.pl' -file './reducer_for_gihyo_2.pl' ?
jobconf mapred.reduce.tasks=12;
hadoop dfs -rmr gihyo3/;
hadoop jar /opt/hadoop/contrib/streaming/hadoop-*-streaming.jar - D mapred.task.timeo
ut=300000 -input gihyo2/ -output gihyo3/ -mapper './mapper_for_gihyo_3.pl'
 -reducer './reducer_for_gihyo_3.pl' -file './mapper_for_gihyo_3.pl' -file './reducer_for_gihyo_3.pl' ?
jobconf mapred.reduce.tasks=12;