Chinaunix首页 | 论坛 | 博客
  • 博客访问: 4370601
  • 博文数量: 291
  • 博客积分: 0
  • 博客等级: 民兵
  • 技术积分: 7924
  • 用 户 组: 普通用户
  • 注册时间: 2016-07-06 14:28
  • 认证徽章:
个人简介

阿里巴巴是个快乐的青年

文章分类

全部博文(291)

文章存档

2018年(21)

2017年(4)

2016年(5)

2015年(17)

2014年(68)

2013年(174)

2012年(2)

分类: HADOOP

2014-09-10 19:14:32

一、原理
        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

        更多详细信息请参看这里,还有这里


阅读(3664) | 评论(1) | 转发(3) |
给主人留下些什么吧!~~

scq2099yt2014-09-10 19:14:51

文明上网,理性发言...

评论热议
请登录后评论。

登录 注册