加载中…
个人资料
enjoydata
enjoydata
  • 博客等级:
  • 博客积分:0
  • 博客访问:395,567
  • 关注人气:29
  • 获赠金笔:0支
  • 赠出金笔:0支
  • 荣誉徽章:
相关博文
推荐博文
谁看过这篇博文
加载中…
正文 字体大小:

[转]hadoop stream 参数详解

(2015-01-07 16:48:43)
标签:

hadoop

stream

参数

it

分类: Hadoop
原文地址:Hadoop streaming作者:tivoli_chen

  1 hadoop streaming

  1. Hadoop streaming是和hadoop一起发布的实用程序。它允许用户创建和执行使用任何程序或者脚本编写的map或者reduce的mapreducejobs。譬如,  
  2.   
  3. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar  
  4.   
  5. -input myInputDirs  
  6.   
  7. -output myOutputDir  
  8.   
  9. -mapper /bin/cat  
  10.   
  11. -reducer /bin/wc  

2hadoop straming工作方式

  1. 在上面的例子中,mapper和reducer都是可执行程序,mapper的输入是读取stdin,reducer的输出是输出到stdout。Hadoopstreaming可以创建mapreduce任务,提交任务到正确的cluster,监控任务的执行过程直到任务完成。  
  2.   
  3. 当mapper定义为可执行程序时,每个mapper task初始化都会独立启动该进程。Mappertask运行时,将输入的文件转换成行来处理,然后传入到stdin;同时,将输出处理为key-value,作为mapper的输出;默认情况下,每行数据第一个tab前面的数据是key,剩下的作为value。如果某行数据中没有tab,则将整行数据作为key,value值为null。然而,这些也可以自定义处理。  
  4.   
  5. 当reducer定义为可执行程序时,每个reducer task初始化都会独立启动该进程。Reducertask运行时,将输入的key-value数据转换成行数据,作为reducer的输入。同时,reducer收集行数据,将行数据转换成key-value形式输出。默认情况下,每行数据第一个tab前面的数据是key,剩下的作为value。然而,这些也可以自定义处理。  
  6.   
  7. 这是mapreduce框架和hadoop streaming之间的基本通信协议。  
  8.   
  9. 用户也可以定义java类作为mapper和reducer,例如,  
  10.   
  11. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar  
  12.   
  13. -input myInputDirs  
  14.   
  15. -output myOutputDir  
  16.   
  17. -mapper org.apache.hadoop.mapred.lib.IdentityMapper  
  18.   
  19. -reducer /bin/wc  
  20.   
  21. 用户可以定义stream.non.zero.exit.is.failure的值为true或者false来表示streaming任务是成功还是失败退出。  

3Job提交设置file选项

  1. 用户可以定义任何程序作为mapper或者reducer。这些可执行程序不需要预先存放在集群的机器上。如果不能,可以通过-file选项来设置程序,提交给job。例如,  
  2.   
  3. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar  
  4.   
  5. -input myInputDirs  
  6.   
  7. -output myOutputDir  
  8.   
  9. -mapper myPythonScript.py  
  10.   
  11. -reducer /bin/wc  
  12.   
  13. -file myPythonScript.py  
  14.   
  15. 该例子将python可执行程序作为mapper。选项-filemyPythonScript.py导致python脚本作为job提交的一部分被传送到集群的机器上。除了可执行程序文件,用户也可以打包一些其他的可能被mapper或者reducer使用的文件。例如,  
  16.   
  17. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar  
  18.   
  19. -input myInputDirs  
  20.   
  21. -output myOutputDir  
  22.   
  23. -mapper myPythonScript.py  
  24.   
  25. -reducer /bin/wc  
  26.   
  27. -file myPythonScript.py  
  28.   
  29. -file myDictionary.txt  

4streaming选项和用法

4.1只有mapper的job

 

  1. 有时候,用户仅仅需要通过mapper函数对输入数据进行处理。要做到这些,可以设置mapred.reduce.tasks=0。这样,mapreduce框架就不会创建reducetask。并且,mapper的输出就是job的最终输出。为了支持向下兼容,hadoop streaming也支持选项-reduceNONE,等价于-D mapred.reduce.tasks=0。  

 

4.2定义jobs的其他选项

  1. 作为一个正常的mapreduce job,用户可以定义hadoop streaming job的其他选项,  
  2.   
  3. -inputformat JavaClassName  
  4.   
  5. -outputformat JavaClassName  
  6.   
  7. -partitioner JavaClassName  
  8.   
  9. -combiner JavaClassName  
  10.   
  11. 设置input format的类应该返回text类的key-value键值对。如果没有设置inputformat类,默认的是TextInputFormat。TextInputFormat返回LongWritable类的keys,它并不是输入数据的一部分,keys可能被舍弃;只有values被传送到streamingmapper中。设置output format的类应该是text类的key-value键值对。如果没有定义outputformat类,默认的是TextOutputFormat。  

4.3 hadoop streaming中的大文件和文件档案

  1. -files 和–archives选项允许用户设置task的文件和文件档案。参数是已经在hdfs上的文件和文件档案的URI。这些文件和文件档案在job中缓存。用户可以从fs.default.name配置变量中获取host和fs_port的值。例如,-files选项,  
  2.   
  3. -files hdfs://host:fs_port/user/testfile.txt#testlink  
  4.   
  5. 在这个例子中,url中#后面的部分是当前工作任务路径的符号链接。因此,这些任务拥有指向本地文件的符号链接。多个输入可以作如下设置,  
  6.   
  7. -files hdfs://host:fs_port/user/testfile1.txt#testlink1  
  8.   
  9. -files hdfs://host:fs_port/user/testfile2.txt#testlink2  
  10.   
  11. -archives选项允许用户复制jar包到当前任务的工作路径,并且自动解压jar包。例如,  
  12.   
  13. -archives hdfs://host:fs_port/user/testfile.jar#testlink3  
  14.   
  15. 在这个例子中,符号链接testlink3创建在当前任务的工作路径中。符号链接指向存放解压jar包的文件路径。  
  16.   
  17. -archives选项的另外一些例子,input.txt文件有两行数据,定义两个文件的名称,testlink/cache.txt和testlink/cache2.txt。testlink指向文件目录的符号链接,拥有两个文件cache.txt 和cache2.txt。  
  18.   
  19. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar  
  20.   
  21. -input"/user/me/samples/cachefile/input.txt"  
  22.   
  23. -mapper "xargs cat"  
  24.   
  25. -reducer "cat"  
  26.   
  27. -output"/user/me/samples/cachefile/out"  
  28.   
  29. -archives  
  30.   
  31. 'hdfs://hadoop-nn1.example.com/user/me/samples/cachefile/cachedir.jar#testlink'  
  32.   
  33. -D mapred.map.tasks=1  
  34.   
  35. -D mapred.reduce.tasks=1  
  36.   
  37. -Dmapred.job.name="Experiment"  
  38.   
  39. ls test_jar/  
  40.   
  41. cache.txt cache2.txt  
  42.   
  43. jar cvf cachedir.jar -C test_jar/  
  44.   
  45. added manifest  
  46.   
  47. adding: cache.txt(in 30(out29)(deflated 3%)  
  48.   
  49. adding: cache2.txt(in 37(out35)(deflated 5%)  
  50.   
  51. hadoop dfs -put cachedir.jar samples/cachefile  
  52.   
  53. hadoop dfs -cat /user/me/samples/cachefile/input.txt  
  54.   
  55. testlink/cache.txt  
  56.   
  57. testlink/cache2.txt  
  58.   
  59. cat test_jar/cache.txt  
  60.   
  61. This is just the cache string  
  62.   
  63. cat test_jar/cache2.txt  
  64.   
  65. This is just the second cache string  
  66.   
  67. hadoop dfs -ls /user/me/samples/cachefile/out  
  68.   
  69. Found items  
  70.   
  71. /user/me/samples/cachefile/out/part-00000 <</span>r3> 69  
  72.   
  73. hadoop dfs -cat /user/me/samples/cachefile/out/part-00000  
  74.   
  75. This is just the cache string  
  76.   
  77. This is just the second cache string  


4.4为jobs定义其他的配置变量

  1. 用户可以使用-D<</span>n>=<</span>v>定义其他配置变量。例如,  
  2.   
  3. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar  
  4.   
  5. -input myInputDirs  
  6.   
  7. -output myOutputDir  
  8.   
  9. -mapper org.apache.hadoop.mapred.lib.IdentityMapper  
  10.   
  11. -reducer /bin/wc  
  12.   
  13. -D mapred.reduce.tasks=2  
  14.   
  15. 选项-D mapred.reduce.tasks=2定义job的reducer为2。  


4.5其他支持的选项

Streaming支持hadoop常用命令行选项。支持的参数主要有下面这些:

bin/hadoop command [genericOptions] [commandOptions]

  1. 改变本地临时文件夹  
  2.   
  3. -D dfs.data.dir=/tmp  
  4.   
  5. 定义其他本地临时文件夹  
  6.   
  7. -D mapred.local.dir=/tmp/local  
  8.   
  9. -D mapred.system.dir=/tmp/system  
  10.   
  11. -D mapred.temp.dir=/tmp/temp  
  12.   
  13. 在streaming命令中设置环境变量  
  14.   
  15. -cmdenv EXAMPLE_DIR=/home/example/dictionaries/  


 

5更多的用法实例

5.1自定义将行数据划分为key-value键值对

  1. 当mapreduce框架从mapper的stdout中读取每行数据时,它将每行数据划分为key-value键值对。默认情况下,每行数据的第一个tab前的数据是key,剩下的是value(除去tab)。  
  2.   
  3. 然而,用户可以自定义该默认设置。用户可以自定义分隔符(除了默认的tab),并且用户也可以定义第n个字符作为kae-value的分隔符。例如,  
  4.   
  5. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar  
  6.   
  7. -input myInputDirs  
  8.   
  9. -output myOutputDir  
  10.   
  11. -mapper org.apache.hadoop.mapred.lib.IdentityMapper  
  12.   
  13. -reducer org.apache.hadoop.mapred.lib.IdentityReducer  
  14.   
  15. -D stream.map.output.field.separator=.  
  16.   
  17. -D stream.num.map.output.key.fields=4  
  18.   
  19. 选项-D stream.map.output.field.separator=.定义mapoutput字段的分隔符为.。第四个.前面的是key,后面的是value。如果该行.的个数少于四个,则整行数据就是key,value是空。  
  20.   
  21. 同样,也可以设置选项-Dstream.reduce.output.field.separator=SEP-Dstream.num.reduce.output.fields=NUM来设置reduce输出的key-value。  

5.2有用的Partitioner类

  1. Hadoop拥有类KeyFieldBasedPartitioner,对很多应用程序有用。该类在某些key字段的基础上允许mapreduce框架划分map的输出。例如,  
  2.   
  3. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar  
  4.   
  5. -input myInputDirs  
  6.   
  7. -output myOutputDir  
  8.   
  9. -mapper org.apache.hadoop.mapred.lib.IdentityMapper  
  10.   
  11. -reducer org.apache.hadoop.mapred.lib.IdentityReducer  
  12.   
  13. -partitionerorg.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner  
  14.   
  15. -D stream.map.output.field.separator=.  
  16.   
  17. -D stream.num.map.output.key.fields=4  
  18.   
  19. -D map.output.key.field.separator=.  
  20.   
  21. -D mapred.text.key.partitioner.options=-k1,2   
  1. 附注:-k1,2 指定对key进行划分后第1 2个域进行划分(上述解释没有找到相关文档,也不属于原文)  
  1. -D mapred.reduce.tasks=12  
  2.   
  3. 例如,  
  4.   
  5. Output输出(keys)  
  6.   
  7. 11.12.1.2  
  8.   
  9. 11.14.2.3  
  10.   
  11. 11.11.4.1  
  12.   
  13. 11.12.1.1  
  14.   
  15. 11.14.2.2  
  16.   
  17. 划分到3个reducer(前面2个字段作为partition的keys)  
  18.   
  19. 11.11.4.1  
  20.   
  21. -----------  
  22.   
  23. 11.12.1.2  
  24.   
  25. 11.12.1.1  
  26.   
  27. -----------  
  28.   
  29. 11.14.2.3  
  30.   
  31. 11.14.2.2  
  32.   
  33. Reducer的每个划分内排序(4个字段同时用于排序)  
  34.   
  35. 11.11.4.1  
  36.   
  37. -----------  
  38.   
  39. 11.12.1.1  
  40.   
  41. 11.12.1.2  
  42.   
  43. -----------  
  44.   
  45. 11.14.2.2  
  46.   
  47. 11.14.2.3  


5.3Comparator类

  1. Hadoop拥有类KeyFieldBasedComparator,在很多程序中得到应用。例如,  
  2.   
  3. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar  
  4.   
  5. -input myInputDirs  
  6.   
  7. -output myOutputDir  
  8.   
  9. -mapper org.apache.hadoop.mapred.lib.IdentityMapper  
  10.   
  11. -reducer org.apache.hadoop.mapred.lib.IdentityReducer  
  12.   
  13. -D  
  14.   
  15. mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator  
  16.   
  17. -D stream.map.output.field.separator=.  
  18.   
  19. -D stream.num.map.output.key.fields=4  
  20.   
  21. -D map.output.key.field.separator=.  
  22.   
  23. -D mapred.text.key.comparator.options=-k2,2nr  
附注:-k2,2nr 中-k2,2指定key分割后的第2个域进行排序,n 指定使用数字排序,r指定排序结果最后要进行反转
  1. -D mapred.reduce.tasks=12  
  2.   
  3. Map输出(keys)  
  4.   
  5. 11.12.1.2  
  6.   
  7. 11.14.2.3  
  8.   
  9. 11.11.4.1  
  10.   
  11. 11.12.1.1  
  12.   
  13. 11.14.2.2  
  14.   
  15. Reducer的输出(使用第二个字段进行排序)  
  16.   
  17. 11.14.2.3  
  18.   
  19. 11.14.2.2  
  20.   
  21. 11.12.1.2  
  22.   
  23. 11.12.1.1  
  24.   
  25. 11.11.4.1  


5.4Hadoop Aggregate包(-reduce aggregate选项)

  1. -D mapred.reduce.tasks=12  
  2.   
  3. Python文件AggregatorForKeyCount.py  
  4.   
  5. #!/usr/bin/python  
  6.   
  7. import sys;  
  8.   
  9. def generateLongCountToken(id):  
  10.         return "LongValueSum:" id +"t" +"1"  
  11.   
  12. def main(argv):  
  13.         line sys.stdin.readline();  
  14.         try:  
  15.                 while line:  
  16.                         line line[:-1];  
  17.                         fields line.split("t");  
  18.                         print generateLongCountToken(fields[0]);  
  19.                         line sys.stdin.readline();  
  20.         except "end of file":  
  21.                 return None  
  22.   
  23. if __name__ == "__main__":  
  24.         main(sys.argv)  

5.5字段选择

  1. Hadoop有类org.apache.hadoop.mapred.lib.FieldSelectionMapReduce。该类允许用户像unix工具中的cut命令来处理文本数据。该类中的Map函数将每个输入的键值对作为字段列表,用户可以自定义字段分隔符。用户可以选择字段列表作为map的输出的key,其他的字段列表作为map的输出的value。Reduce函数与此类似。  
  2.   
  3. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar  
  4.   
  5. -input myInputDirs  
  6.   
  7. -output myOutputDir  
  8.   
  9. -mapperorg.apache.hadoop.mapred.lib.FieldSelectionMapReduce  
  10.   
  11. -reducerorg.apache.hadoop.mapred.lib.FieldSelectionMapReduce  
  12.   
  13. -partitionerorg.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner  
  14.   
  15. -D map.output.key.field.separa=.  
  16.   
  17. -D mapred.text.key.partitioner.options=-k1,2  
  18.   
  19. -D mapred.data.field.separator=.  
  20.   
  21. -D map.output.key.value.fields.spec=6,5,1-3:0-  
  22.   
  23. -D reduce.output.key.value.fields.spec=0-2:5-  
  24.   
  25. -D mapred.reduce.tasks=12  

5.6 mapred尝试任务失败次数控制及map任务失败率控制

  1. -D mapred.map.max.attempts="3"                                                                                    
  2. -D mapred.reduce.max.attempts="3"                                                                                 
  3. -D mapred.max.map.failures.percent="1"  设置map任务失败率容忍率  

5.7 mapred限制java读取数据行的最大长度(防止mapred程序执行时进度停滞且报heatbeat错误):
  1. -D  mapred.linerecordreader.maxlength 409600  

0

阅读 评论 收藏 转载 喜欢 打印举报/Report
  • 评论加载中,请稍候...
发评论

    发评论

    以上网友发言只代表其个人观点,不代表新浪网的观点或立场。

      

    新浪BLOG意见反馈留言板 电话:4000520066 提示音后按1键(按当地市话标准计费) 欢迎批评指正

    新浪简介 | About Sina | 广告服务 | 联系我们 | 招聘信息 | 网站律师 | SINA English | 会员注册 | 产品答疑

    新浪公司 版权所有