Hadoop streaming裡的key和value格式不一樣，實際上雖然也是以K1，V1輸入輸出的，但是是以line的格式輸出的。因此中間經過partition過後的K2，V2S的那個iterable的函式格式貌似是沒有的！

比如，我想寫一個wordcount

在Python中，標準輸入輸出格式是：

import  sys
for line in sys.stdin:

每一行是一個輸入，然後進入之後的K1，V1你自己切分，自己搞成想要的K2，V2之後，print出來！！
print的格式是print K2+”\t”+V2
這樣print出來的格式實際上是print了一行【一個line】，系統會自動預設識別第一個\t 把前面作為key自動給partition了
然後reducer的時候，你同樣要
for line in sys.stdin:
然後把partition過後的每一個line收進來。自己處理。

附上一段轉自別人部落格的話：
Hadoop Streaming程式設計
一、概述
Hadoop Streaming是Hadoop提供的一個程式設計工具，它允許使用者使用任何可執行檔案或者指令碼作為Mapper和Reducer，例如：
採用shell指令碼語言中的一些命令作為mapper和reducer（cat作為mapper，wc作為reduce）

$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar\
-input myInputDir\
-output myOutputDir\
-mapper cat\
-reducer wc

本文安排如下，第二節介紹Hadoop Streaming的原理，第三節介紹Hadoop Streaming的使用方法，第四節介紹Hadoop Streaming的程式編寫方法，在這一節中，使用C++、C、Shell指令碼和python實現了WordCount作業，第五節總結了常見的問題。
 

二、Hadoop Streaming原理
mapper和reducer會從標準輸入中讀取使用者資料，一行一行處理後傳送給標準輸出。Streaming工具會建立MapReduce作業，傳送給各個tasktracker，同時監控整個作業的執行過程。
如果一個檔案（可執行或者指令碼）作為mapper，mapper初始化時，每一個mapper任務會把該檔案作為一個單獨程序啟動，mapper任務執行時，它把輸入切分成行，並把每一行提供給可執行檔案程序的標準輸入。同時，mapper收集可執行檔案程序標準輸出的內容，並把收到的每一行內容轉化為key/value對，作為mapper的輸出。預設情況下，一行中第一個tab之前的部分作為key，之後的（不包括tab）作為value。如果沒有tab，整行作為key值，value值為null。
對於reducer，類似。
以上是Map/Reduce框架和Streaming Mapper/Reduce之間的基本通訊協議。

三、Hadoop Streaming用法

四、Mapper和Reducer實現
本節試圖用盡可能多的語言編寫Mapper和Reducer，包括java、C、C++、shell指令碼、python等。
由於hadoop會自動解析資料檔案到Mapper或者Reducer的標準輸入中，以供它們讀取使用，所以應先了解各個語言獲取標準輸入的方法。

現在寫一個wordcount：目的是要統計adult=1的檔案裡，各自文字出現的頻率。
輸入line格式如下所示：
u’20111725 13,模特表演,專業課 獨家,青春,模特,時尚,舞臺 1’
最前面是ID,後面接一個tab鍵，然後是標籤，然後再接一個tab鍵；然後是標題名稱的切分，最後是輸出判定：為1。

因為要做文字分類詞向量模型，所以掐頭去尾，只留中間標籤。

mapper：

#!/usr/bin/python
#-*-coding:utf-8-*-
#!Author:[email protected]
# Modified: 20160615
import sys
# a=[u'20111725        13,模特表演,專業課      獨家,青春,模特,時尚,舞臺        1']
for line in sys.stdin:
    line=line.strip().split()
    # for w in sp:
    #     print w
    # key=sp[0]
    value=line[1:-1]
    value_str=",".join(value)
    value_list=value_str.split(",")
    for key1 in value_list:
        print key1+"\t"+str(1)

打印出的類似於

接下來用 key + \t + 1的每一個line當做輸出，print之後，經過partition之後，把key相同的自動被放到同一個reducer上

reducer程式碼：

test_list=[u'男人 1',u'男人 1',u'潑水節    1',u'潑水節    1',u'潑水節    1',u'潑水節    1',u'潑水節    1',u'猥褻 1']
# import os
# import sys
# sys.path.append("./")
# sys.path.append("/data/programs/common/python/")
# reload(sys)
word2count_dict={}
for line in test_list:
    line=line.strip()
    word,count=line.split()
    try:
        count=int(count)
        word2count_dict[word]=word2count_dict.get(word,0)+count
    except ValueError:
        pass
sorted_word2count_list=sorted(word2count_dict.iteritems(),key=lambda d:d[1],reverse=True)
for word_count_tuple in sorted_word2count_list:
    print word_count_tuple[0]+"\t"+str(word_count_tuple[1])

這裡有兩個注意點：1.關於字典word2count_dict的值提取過程。經常情況下，當沒有該key的時候，用這種方式總會報錯。現在，用.get函式，並設定一個default值，問題得到了解決
get函式的定義如下：

第二，得到之後要對dict里根據值來進行排序。

字典排序

方式有兩種，第一種是用operation類，第二種是不用import，直接做，但是這種方式是根據tuple來進行排序的
sorted(word2count_dict.iteritems(),key=lambda d:d[1],reverse=True)
.iteritems()得到一個(key,value)的tuple，key=d[0]說明是key，d[1]說明是value。預設從小到大，因此用reverse

寫完了mapper和reducer，接下來開始寫主要的那個.sh檔案。

.sh 顧名思義是shell檔案。
最後reducer裡得到的是這樣的檔案值：

之後，寫完mapper和reducer，需要寫sh檔案。

在sh檔案中，需要設定basedir,input_dir, output_dir