的Python的MapReduce：從映射器

Question

join2_mapper.py

#!/usr/bin/env python 
import sys 
shows = [] 
for line in sys.stdin: 
    line  = line.strip() 
    key_value = line.split(',') 

    if key_value[1] == 'ABC': 
     if key_value[1] not in shows: 
      shows.append(key_value[0]) 

    if key_value[1].isdigit() and (key_value[0] in shows): 
     print('{0}\t{1}'.format(key_value[0], key_value[1])) 

樣品I/P空文件

Hourly_Sports,DEF 
Baked_Games,ABC 
Dumb_Talking,ABC 
Surreal_Talking,DEF 
Cold_Sports,BAT 
Hourly_Talking,XYZ 
Baked_Talking,CNO 
PostModern_Games,ABC 
Loud_Talking,DEF 
Almost_News,BAT 
Hot_Talking,XYZ 
Dumb_News,CNO 
Surreal_News,ABC 
Cold_Talking,DEF 
Hourly_Show,BAT 
Baked_Show,XYZ 
PostModern_Talking,CNO 
Loud_Show,ABC 
Almost_Cooking,DEF 
Hot_News,BAT 
Dumb_Cooking,XYZ 
Surreal_Cooking,CNO 
Cold_News,ABC 
Hourly_Sports,DEF 
Baked_Sports,BAT 
PostModern_Show,XYZ 
Loud_Sports,CNO 
Almost_Games,ABC 
Hot_Cooking,DEF 
Dumb_Games,BAT 
Surreal_Games,XYZ 
Cold_Cooking,CNO 
Hourly_Talking,ABC 
Baked_Talking,DEF 
PostModern_Sports,BAT 
Loud_Talking,XYZ 
Almost_Talking,CNO 
Hot_Games,ABC 
Dumb_Talking,DEF 
Surreal_Talking,BAT 
Cold_Games,XYZ 
Hourly_News,CNO 
Baked_News,ABC 
PostModern_Talking,DEF 
Loud_News,BAT 
Almost_Show,XYZ 
Hot_Talking,CNO 
Dumb_Show,ABC 
Surreal_Show,DEF 
Cold_Talking,BAT 
Hourly_Cooking,XYZ 
Baked_Cooking,CNO 
PostModern_News,ABC 
Loud_Cooking,DEF 
Almost_Sports,BAT 
Hot_Show,XYZ 
Dumb_Sports,CNO 
Surreal_Sports,ABC 
Cold_Show,DEF 
Hourly_Games,BAT 
Baked_Games,XYZ 
PostModern_Cooking,CNO 
Loud_Games,ABC 
Almost_Talking,DEF 
Hot_Sports,BAT 
Dumb_Talking,XYZ 
Surreal_Talking,CNO 
Cold_Sports,ABC 
Hourly_Talking,DEF 
Baked_Talking,BAT 
PostModern_Games,XYZ 
Loud_Talking,CNO 
Almost_News,ABC 
Hot_Talking,DEF 
Dumb_News,BAT 
Surreal_News,XYZ 
Cold_Talking,CNO 
Hourly_Show,ABC 
Almost_Cooking,855 
Baked_Games,991 
Baked_News,579 
Baked_Games,200 
Baked_Games,533 
Cold_News,590 
Hourly_Show,896 

$ cat j2.txt | python join2_mapper.py

Baked_Games 991 
Baked_News 579 
Baked_Games 200 
Baked_Games 533 
Cold_News 590 
Hourly_Show 896 

hadoop jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar -input /user/cloudera/join2_data/join2_genchan*.txt -input /user/cloudera/join2_data/join2_gennum*.txt -output /user/cloudera/join2_f1f -mapper /home/cloudera/join2_mapper.py -reducer /home/cloudera/join2_reducer.py -numReduceTasks 0

Map-Reduce Framework 
     Map input records=6600 
     Map output records=0 
     Input split bytes=759 
     Spilled Records=0 
     Failed Shuffles=0 
     Merged Map outputs=0 
     GC time elapsed (ms)=4419 
     CPU time spent (ms)=9170 
     Physical memory (bytes) snapshot=702300160 
     Virtual memory (bytes) snapshot=9022578688 
     Total committed heap usage (bytes)=364511232 
    File Input Format Counters 
     Bytes Read=113055 
    File Output Format Counters 
     Bytes Written=0 

問題在於輸入文件。其實我也六個輸入文件內容如下：

$ hdfs dfs -ls /user/cloudera/join2_data/join2_gen*.txt 
-rw-r--r-- 1 cloudera cloudera  1714 2015-11-07 12:24 /user/cloudera/join2_data/join2_genchanA.txt 
-rw-r--r-- 1 cloudera cloudera  3430 2015-11-07 12:24 /user/cloudera/join2_data/join2_genchanB.txt 
-rw-r--r-- 1 cloudera cloudera  5152 2015-11-07 12:24 /user/cloudera/join2_data/join2_genchanC.txt 
-rw-r--r-- 1 cloudera cloudera  17114 2015-11-07 12:24 /user/cloudera/join2_data/join2_gennumA.txt 
-rw-r--r-- 1 cloudera cloudera  34245 2015-11-07 12:24 /user/cloudera/join2_data/join2_gennumB.txt 
-rw-r--r-- 1 cloudera cloudera  51400 2015-11-07 12:24 /user/cloudera/join2_data/join2_gennumC.txt 

當我所有的文件連接成一個單一的文件，並運行它正在工作。獲得理想的結果。當提供六個塊的輸入文件時，我會得到一個空文件。請指教。

Answer 1

只提供一個-input參數，並將其傳遞到包含所有輸入數據的文件夾的路徑，而不是使用正則表達式。如果您不使用減速機，請拆下減速機（只是爲了消除雜亂）。我不能確切地說哪個能解決這個問題（我懷疑它是第一個），但它會解決這個問題。所以：

hadoop jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar \ 
-input /user/cloudera/join2_data/ \ 
-output /user/cloudera/join2_f1f \ 
-mapper /home/cloudera/join2_mapper.py 

Answer 2

沒你的意思是有key_value[0]，而不是1，在if key_value[1] not in shows？