Hadoop Streaming 是Hadoop提供方的一個程式設計工具，它允許使用者使用任何可執行檔案或者指令碼作為Mapper和Reducer。

Hadoop Streaming 多語言程式設計

1. 以標準輸入流作為輸入：

    1) C++: cin

    2) Php: stdin

    3) Python: sys.stdin

2. 以標準的輸出流作為輸出：

    1) C++: cout

    2) Php: echo

    3) Python: print

3. 可實現Mapper和Reducer，其他元件（InputFormat，Partitioner等需要用Java語言來實現）

Python實現Wordcount：

1. mapper.py

[[email protected] wordcount]# vim mapper.py

寫入

#!/usr/bin/python
import sys
word2count = {}
for line in sys.stdin:
    line = line.strip()
    words = filter(lambda word:word,line.split())
    for word in words:
        print("%s\t%s" % (word,1))

2. reducer.py

[[email protected] wordcount]# vim reducer.py
 

#!/usr/bin/python
from operator import itemgetter
import sys

word2count = {}
for line in sys.stdin:
    line = line.strip()
    word,count = line.split()
    try:
        count = int(count)
        word2count[word] = word2count.get(word,0) + count
    except ValueError as err:
        print(err)
        pass

sorted_word2count = sorted(word2count.items(),key=itemgetter(0))
for word,count in sorted_word2count:
    print("%s\t%s" % (word, count))
 

3. 準備一個測試檔案test.txt

[[email protected] wordcount]# vim test.txt

this is a test
this is a test
this is a test
this is a test

4. 本地測試

[[email protected] wordcount]# cat test.txt |python mapper.py |sort|python reducer.py 
a	4
is	4
test	4
this	4
[[email protected] wordcount]#

5. 叢集執行

[[email protected] wordcount]# hadoop fs -mkdir /user/root/wordcount
[[email protected] wordcount]# hadoop fs -put test.txt /user/root/wordcount/
[[email protected] wordcount]# hadoop fs -ls /user/root/wordcount/
Found 1 items
-rw-r--r--   3 root root         60 2018-05-14 09:58 /user/root/wordcount/test.txt
[[email protected] wordcount]# 

執行mapreduce

[[email protected] wordcount]# hadoop jar /opt/cloudera/parcels/CDH-5.13.1-1.cdh5.13.1.p0.2/jars/hadoop-streaming-2.6.0-cdh5.13.1.jar -D mapred.reduce.tasks=1 -mapper "python mapper.py" -reducer "python reducer.py" -file mapper.py -file reducer.py -input /user/root/wordcount/test.txt -output /user/root/wordcount/out
18/05/14 10:00:37 WARN streaming.StreamJob: -file option is deprecated, please use generic option -files instead.
packageJobJar: [mapper.py, reducer.py] [/opt/cloudera/parcels/CDH-5.13.1-1.cdh5.13.1.p0.2/jars/hadoop-streaming-2.6.0-cdh5.13.1.jar] /tmp/streamjob7327942722840197442.jar tmpDir=null
18/05/14 10:00:39 INFO mapred.FileInputFormat: Total input paths to process : 1
18/05/14 10:00:39 INFO mapreduce.JobSubmitter: number of splits:2
18/05/14 10:00:39 INFO Configuration.deprecation: mapred.reduce.tasks is deprecated. Instead, use mapreduce.job.reduces
18/05/14 10:00:40 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1526119303078_0004
18/05/14 10:00:40 INFO impl.YarnClientImpl: Submitted application application_1526119303078_0004
18/05/14 10:00:40 INFO mapreduce.Job: The url to track the job: http://g13-1.novalocal:8088/proxy/application_1526119303078_0004/
18/05/14 10:00:40 INFO mapreduce.Job: Running job: job_1526119303078_0004
18/05/14 10:00:46 INFO mapreduce.Job: Job job_1526119303078_0004 running in uber mode : false
18/05/14 10:00:46 INFO mapreduce.Job:  map 0% reduce 0%
18/05/14 10:00:50 INFO mapreduce.Job:  map 100% reduce 0%
18/05/14 10:00:56 INFO mapreduce.Job:  map 100% reduce 100%
18/05/14 10:00:57 INFO mapreduce.Job: Job job_1526119303078_0004 completed successfully
18/05/14 10:00:57 INFO mapreduce.Job: Counters: 49
	File System Counters
		FILE: Number of bytes read=61
		FILE: Number of bytes written=469610
		FILE: Number of read operations=0
		FILE: Number of large read operations=0
		FILE: Number of write operations=0
		HDFS: Number of bytes read=290
		HDFS: Number of bytes written=23
		HDFS: Number of read operations=9
		HDFS: Number of large read operations=0
		HDFS: Number of write operations=2
	Job Counters 
		Launched map tasks=2
		Launched reduce tasks=1
		Rack-local map tasks=2
		Total time spent by all maps in occupied slots (ms)=5834
		Total time spent by all reduces in occupied slots (ms)=3080
		Total time spent by all map tasks (ms)=5834
		Total time spent by all reduce tasks (ms)=3080
		Total vcore-milliseconds taken by all map tasks=5834
		Total vcore-milliseconds taken by all reduce tasks=3080
		Total megabyte-milliseconds taken by all map tasks=5974016
		Total megabyte-milliseconds taken by all reduce tasks=3153920
	Map-Reduce Framework
		Map input records=4
		Map output records=16
		Map output bytes=92
		Map output materialized bytes=104
		Input split bytes=200
		Combine input records=0
		Combine output records=0
		Reduce input groups=4
		Reduce shuffle bytes=104
		Reduce input records=16
		Reduce output records=4
		Spilled Records=32
		Shuffled Maps =2
		Failed Shuffles=0
		Merged Map outputs=2
		GC time elapsed (ms)=235
		CPU time spent (ms)=3290
		Physical memory (bytes) snapshot=1292472320
		Virtual memory (bytes) snapshot=8453484544
		Total committed heap usage (bytes)=1889533952
	Shuffle Errors
		BAD_ID=0
		CONNECTION=0
		IO_ERROR=0
		WRONG_LENGTH=0
		WRONG_MAP=0
		WRONG_REDUCE=0
	File Input Format Counters 
		Bytes Read=90
	File Output Format Counters 
		Bytes Written=23
18/05/14 10:00:57 INFO streaming.StreamJob: Output directory: /user/root/wordcount/out
[[email protected] wordcount]# 

命令列檢視結果

[[email protected] wordcount]# hadoop fs -cat /user/root/wordcount/out/part-00000
a	4
is	4
test	4
this	4
[[email protected] wordcount]# 

官方文件：https://hadoop.apache.org/docs/r1.0.4/cn/streaming.html