Hadoop streaming
Hadoop为MapReduce提供了不同的API,可以方便我们使用不同的编程语言来使用MapReduce框架,而不是只局限于Java。这里要介绍的就是Hadoop streaming API。Hadoop streaming 使用Unix的standard streams作为我们mapreduce程序和MapReduce框架之间的接口。所以你可以用任何语言来编写MapReduce程序,只要该语言可以往standard input/output上进行读写。
streamming是天然适用于文字处理的(text processing),当然,也仅适用纯文本的处理,对于需要对象和序列化的场景,hadoop streaming无能为力。它力图使我们能够快捷的通过各种脚本语言,快速的处理大量的文本文件。以下是steaming的一些特点:
- Map函数的输入是通过stand input一行一行的接收数据的。(不像Java API,通过InputFormat类做预处理,使得Map函数的输入是有Key和value的)
- Map函数的output则必须限定为key-value pair,key和value之间用\t分开。(MapReduce框架在处理intermediate的Map输出时,必须做sort和partition,即shuffle)
- Reduce函数的input是Map函数的output也是key-value pair,key和value之间用\t分开。
常用的Streaming编程语言:
- bash shell
- ruby
- python
Ruby
下面是一个Ruby编写的MapReduce程序的示例:
map
max_temperature_map.rb:
ruby #!/usr/bin/env ruby STDIN.each_line do |line| val = line year, temp, q = val[15,4], val[87,5], val[92,1] puts "#{year}\t#{temp}" if (temp != "+9999" && q =~ /[01459]/) end
- 从标准输入读入一行data。
- 处理数据之后,生成一个键值对,用\t分隔,输出到标准输出
reduce
max_temperature_reduce.rb:
ruby #!/usr/bin/env ruby last_key, max_val = nil, -1000000 STDIN.each_line do |line| key, val = line.split("\t") if last_key && last_key != key puts "#{last_key}\t#{max_val}" last_key, max_val = key, val.to_i else last_key, max_val = key, [max_val, val.to_i].max end end puts "#{last_key}\t#{max_val}" if last_key
- 从标准输入读入一行数据
- 数据是用\t分隔的键值对
- 数据是被MapReduce根据key排序之后顺序一行一行读入
- reduce函数对数据进行处理,并输出,输出仍是用\t分隔的键值对
运行
% hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar -input input/ncdc/sample.txt -output output -mapper ch02/src/main/ruby/max_temperature_map.rb -reducer ch02/src/main/ruby/max_temperature_reduce.rb
- hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar指明了使用hadoop streaming
- hadoop-*-streaming.jar会将input里的文件,一行一行的输出到标准输出。
- 用-mapper指定Map函数。类似于通过管道将数据传给rb文件: data|ch02/src/main/ruby/max_temperature_map.rb
- -reducer指定Reduce函数。
Python
Map
#!/usr/bin/env python import re import sys for line in sys.stdin: val = line.strip() (year, temp, q) = (val[15:19], val[87:92], val[92:93]) if (temp != "+9999" and re.match("[01459]", q)): print "%s\t%s" % (year, temp)
Reduce
#!/usr/bin/env python import sys (last_key, max_val) = (None, -sys.maxint) for line in sys.stdin: (key, val) = line.strip().split("\t") if last_key and last_key != key: print "%s\t%s" % (last_key, max_val) (last_key, max_val) = (key, int(val)) else: (last_key, max_val) = (key, max(max_val, int(val))) if last_key: print "%s\t%s" % (last_key, max_val)
运行
% hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar -input input/ncdc/sample.txt -output output -mapper ch02/src/main/ruby/max_temperature_map.py-reducer ch02/src/main/ruby/max_temperature_reduce.py
Bash shell
Map
#!/usr/bin/env bash # NLineInputFormat gives a single line: key is offset, value is S3 URI read offset s3file # Retrieve file from S3 to local disk echo "reporter:status:Retrieving $s3file" >&2 $HADOOP_INSTALL/bin/hadoop fs -get $s3file . # Un-bzip and un-tar the local file target=`basename $s3file .tar.bz2` mkdir -p $target echo "reporter:status:Un-tarring $s3file to $target" >&2 tar jxf `basename $s3file` -C $target # Un-gzip each station file and concat into one file echo "reporter:status:Un-gzipping $target" >&2 for file in $target/*/* do gunzip -c $file $target.all echo "reporter:status:Processed $file" >&2 done # Put gzipped version into HDFS echo "reporter:status:Gzipping $target and putting in HDFS" >&2 gzip -c $target.all | $HADOOP_INSTALL/bin/hadoop fs -put - gz/$target.gz
运行
% hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar -D mapred.reduce.tasks=0 -D mapred.map.tasks.speculative.execution=false -D mapred.task.timeout=12000000 -input ncdc_files.txt -inputformat org.apache.hadoop.mapred.lib.NLineInputFormat -output output -mapper load_ncdc_map.sh -file load_ncdc_map.sh
- 这里的-D mapred.reduce.tasks=0将reduce task观掉,因此也不需要设置-reducer
- 只使用Mapper,可以通过MapReduce帮助我们并行的完成一些平时只能串行的shell脚本
- 注意这里的-file,在集群模式下,需要并行运行时,需要-file把文件传输到其他节点
Combiner
在streaming模式下,仍然可以运行Combiner,两种方法:
- 通过Java编写一个combiner的函数,并使用-combiner option
- 以命令行的管道模式完成combiner的任务
这里具体解释第二种方法:
% hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar -input input/ncdc/all -output output -mapper "ch02/src/main/ruby/max_temperature_map.rb | sort | ch02/src/main/ruby/max_temperature_reduce.rb" -reducer ch02/src/main/ruby/max_temperature_reduce.rb -file ch02/src/main/ruby/max_temperature_map.rb -file ch02/src/main/ruby/max_temperature_reduce.rb
注意看-mapper这一行,通关管道的方式,把mapper的临时输出文件(intermediate file,Map完成后的临时文件)作为输入,送到sort进行排序,然后送到reduce脚本,来完成类似于combiner的工作。这时候的输出才真正的作为shuffle的输入,被分组并在网络上发送到Reduce
感谢阅读,希望能帮助到大家,谢谢大家对本站的支持!
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
稳了!魔兽国服回归的3条重磅消息!官宣时间再确认!
昨天有一位朋友在大神群里分享,自己亚服账号被封号之后居然弹出了国服的封号信息对话框。
这里面让他访问的是一个国服的战网网址,com.cn和后面的zh都非常明白地表明这就是国服战网。
而他在复制这个网址并且进行登录之后,确实是网易的网址,也就是我们熟悉的停服之后国服发布的暴雪游戏产品运营到期开放退款的说明。这是一件比较奇怪的事情,因为以前都没有出现这样的情况,现在突然提示跳转到国服战网的网址,是不是说明了简体中文客户端已经开始进行更新了呢?