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
感谢阅读，希望能帮助到大家，谢谢大家对本站的支持！
《魔兽世界》大逃杀！60人新游玩模式《强袭风暴》3月21日上线
暴雪近日发布了《魔兽世界》10.2.6 更新内容，新游玩模式《强袭风暴》即将于3月21 日在亚服上线，届时玩家将前往阿拉希高地展开一场 60 人大逃杀对战。
艾泽拉斯的冒险者已经征服了艾泽拉斯的大地及遥远的彼岸。他们在对抗世界上最致命的敌人时展现出过人的手腕，并且成功阻止终结宇宙等级的威胁。当他们在为即将于《魔兽世界》资料片《地心之战》中来袭的萨拉塔斯势力做战斗准备时，他们还需要在熟悉的阿拉希高地面对一个全新的敌人──那就是彼此。在《巨龙崛起》10.2.6 更新的《强袭风暴》中，玩家将会进入一个全新的海盗主题大逃杀式限时活动，其中包含极高的风险和史诗级的奖励。
《强袭风暴》不是普通的战场，作为一个独立于主游戏之外的活动，玩家可以用大逃杀的风格来体验《魔兽世界》，不分职业、不分装备（除了你在赛局中捡到的），光是技巧和战略的强弱之分就能决定出谁才是能坚持到最后的赢家。本次活动将会开放单人和双人模式，玩家在加入海盗主题的预赛大厅区域前，可以从强袭风暴角色画面新增好友。游玩游戏将可以累计名望轨迹，《巨龙崛起》和《魔兽世界：巫妖王之怒 经典版》的玩家都可以获得奖励。
更新动态
- 凤飞飞《我们的主题曲》飞跃制作[正版原抓WAV+CUE]
- 刘嘉亮《亮情歌2》[WAV+CUE][1G]
- 红馆40·谭咏麟《歌者恋歌浓情30年演唱会》3CD[低速原抓WAV+CUE][1.8G]
- 刘纬武《睡眠宝宝竖琴童谣 吉卜力工作室 白噪音安抚》[320K/MP3][193.25MB]
- 【轻音乐】曼托凡尼乐团《精选辑》2CD.1998[FLAC+CUE整轨]
- 邝美云《心中有爱》1989年香港DMIJP版1MTO东芝首版[WAV+CUE]
- 群星《情叹-发烧女声DSD》天籁女声发烧碟[WAV+CUE]
- 刘纬武《睡眠宝宝竖琴童谣 吉卜力工作室 白噪音安抚》[FLAC/分轨][748.03MB]
- 理想混蛋《Origin Sessions》[320K/MP3][37.47MB]
- 公馆青少年《我其实一点都不酷》[320K/MP3][78.78MB]
- 群星《情叹-发烧男声DSD》最值得珍藏的完美男声[WAV+CUE]
- 群星《国韵飘香·贵妃醉酒HQCD黑胶王》2CD[WAV]
- 卫兰《DAUGHTER》【低速原抓WAV+CUE】
- 公馆青少年《我其实一点都不酷》[FLAC/分轨][398.22MB]
- ZWEI《迟暮的花 (Explicit)》[320K/MP3][57.16MB]