Hadoop初探之Stream

一、原理
        Hadoop Streaming是Hadoop提供的一个编程工具，它允许用户使用任何可执行文件或者脚本文件作为Mapper和Reducer，例如：采用shell脚本语言中的一些命令作为mapper和reducer（cat作为mapper，wc作为reducer）
        $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar \
            -input myInputDirs \
            -output myOutputDir \
            -mapper cat \
            -reducer wc
        mapper和reducer会从标准输入中读取用户数据，一行一行处理后发送给标准输出。Streaming工具会创建MapReduce作业，发送给各个tasktracker，同时监控整个作业的执行过程。
        如果一个文件（可执行或者脚本）作为mapper，则在mapper初始化时，每一个mapper任务会把该文件作为一个单独进程启动，mapper任务运行时，它把输入切分成行并把每一行提供给可执行文件进程的标准输入。 同时，mapper收集可执行文件进程标准输出的内容，并把收到的每一行内容转化成key/value对，作为mapper的输出。默认情况下，一行中第一个tab之前的部分作为key，之后的（不包括tab）作为value。如果没有tab，整行作为key值，value值为null。不过，这可以定制，在下文中会介绍如何自定义key和value的切分方式。
        对于reducer，类似。
        以上是Map/Reduce框架和streaming mapper/reducer之间的基本通信协议。

二、语法
1、基本语法
        Usage: $HADOOP_HOME/bin/hadoop jar \
                       $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar [options]
        options：
        （1）-input：输入文件路径
        （2）-output：输出文件路径
        （3）-mapper：用户自己写的mapper程序，可以是可执行文件或者脚本
        （4）-reducer：用户自己写的reducer程序，可以是可执行文件或者脚本
        （5）-file：打包文件到提交的作业中，可以是mapper或者reducer要用的输入文件，如配置文件，字典等。
        （6）-partitioner：用户自定义的partitioner程序
        （7）-combiner：用户自定义的combiner程序（必须用java实现）
        （8）-D：作业的一些属性（以前用的是-jonconf），具体有：
                1）mapred.map.tasks：map task数目
                2）mapred.reduce.tasks：reduce task数目
                3）stream.map.input.field.separator/stream.map.output.field.separator：map task输入/输出数据的分隔符，默认均为\t。
                4）stream.num.map.output.key.fields：指定map task输出记录中key所占的域数目
                5）stream.reduce.input.field.separator/stream.reduce.output.field.separator：reduce task输入/输出数据的分隔符，默认均为\t。
                6）stream.num.reduce.output.key.fields：指定reduce task输出记录中key所占的域数目。
        有时只需要map函数处理输入数据，这时只需把mapred.reduce.tasks设置为零，Map/Reduce框架就不会创建reducer任务，mapper任务的输出就是整个作业的最终输出。为了做到向下兼容，Hadoop Streaming也支持“-reduce None”选项，它与“-jobconf mapred.reduce.tasks=0”等价。
2、扩展语法
        之前已经提到，当Map/Reduce框架从mapper的标准输入读取一行时，它把这一行切分为key/value对。在默认情况下，每行第一个tab符之前的部分作为key，之后的部分作为value（不包括tab符）。
        但是，用户也可以自定义，可以指定分隔符是其它字符而不是默认的tab符，或者指定在第n（n>=1）个分割符处分割而不是默认的第一个。例如：
        $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
            -input myInputDirs \
            -output myOutputDir \
            -mapper org.apache.hadoop.mapred.lib.IdentityMapper \
            -reducer org.apache.hadoop.mapred.lib.IdentityReducer \
            -jobconf stream.map.output.field.separator=. \
            -jobconf stream.num.map.output.key.fields=4 
        在上面的例子中，“-jobconf stream.map.output.field.separator=.”指定“.”作为map输出内容的分隔符，并且从在第4个“.”之前的部分作为key，之后的部分作为value（不包括这第4个“.”）。 如果一行中的“.”少于4个，则整行的内容作为key，value设为空的Text对象（就像这样创建了一个Text：new Text("")）。
        同样地，用户也可以使用“-jobconf stream.reduce.output.field.separator=SEP”和“-jobconf stream.num.reduce.output.fields=NUM”来指定reduce输出的行中，第几个分隔符处分割key和value。

三、实例
    为了说明各种语言编写Hadoop Streaming程序的方法，下面以WordCount为例，WordCount作业的主要功能是对用户输入的数据中所有字符串进行计数。
1、shell
        #vi mapper.sh
        #! /bin/bash
        while read LINE; do
          for word in $LINE
          do
            echo "$word 1"
          done
        done
        -------------------------------------------------------------------------
        #vi reducer.sh
        #! /bin/bash
        count=0
        started=0
        word=""
        while read LINE;do
            newword=`echo $LINE | cut -d ' ' -f 1`
            if [ "$word" != "$newword" ];then
                [ $started -ne 0 ] && echo -e "$word\t$count"
                word=$newword
                count=1
                started=1
            else
                count=$(( $count + 1 ))
            fi
        done
        echo -e "$word\t$count"
        -------------------------------------------------------------------------
        本地测试：cat input.txt | sh mapper.sh | sort | sh reducer.sh
        集群测试：
                $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar \
                    -input myInputDirs \
                    -output myOutputDir \
                    -mapper mapper.sh\
                    -reducer reducer.sh
        如果执行上面脚本提示：“Caused by: java.io.IOException: Cannot run program “/user/hadoop/Mapper”: error=2, No such file or directory”，则说明找不到可执行程序，可以在提交作业时，采用-file选项指定这些文件，比如上面例子中，可以使用“-file mapper.py -file reducer.py”，这样，Hadoop会将这两个文件自动分发到各个节点上，比如：
        $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar \
            -input myInputDirs \
            -output myOutputDir \
            -mapper mapper.sh\
            -reducer reducer.sh\
            -file mapper.sh \
            -file reducer.sh
2、python
        #vi mapper.py
        #!/usr/bin/env python 
        import sys
        #maps words to their counts
        word2count = {}
        #input comes from STDIN (standard input)
        for line in sys.stdin:
            #remove leading and trailing whitespace
            line = line.strip()
            #split the line into words while removing any empty strings
            words = filter(lambda word: word, line.split())
            #increase counters
            for word in words:
                #write the results to STDOUT (standard output);
                #what we output here will be the input for the
                #Reduce step, i.e. the input for reducer.py
                #
                #tab-delimited; the trivial word count is 1
                print '%s\t%s' % (word, 1)
        -------------------------------------------------------------------------
        #vi reducer.py
        #!/usr/bin/env python
        from operator import itemgetter
        import sys
        #maps words to their counts
        word2count = {}
        #input comes from STDIN
        for line in sys.stdin:
            #remove leading and trailing whitespace
            line = line.strip()
            #parse the input we got from mapper.py
            word, count = line.split()
            #convert count (currently a string) to int
            try:
                count = int(count)
                word2count[word] = word2count.get(word, 0) + count
            except ValueError:
                #count was not a number, so silently
                #ignore/discard this line
                pass
        #sort the words lexigraphically;
        #
        #this step is NOT required, we just do it so that our
        #final output will look more like the official Hadoop
        #word count examples
        sorted_word2count = sorted(word2count.items(), key=itemgetter(0))
        #write the results to STDOUT (standard output)
        for word, count in sorted_word2count:
            print '%s\t%s'% (word, count)
        -------------------------------------------------------------------------
        本地测试：
                cat input.txt | python mapper.py | sort | python reducer.py
        集群测试：
                $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar \
                    -input myInputDirs \
                    -output myOutputDir \
                    -mapper mapper.py\
                    -reducer reducer.py

        更多详细信息请参看，还有。