Hadoop streaming详细介绍

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的一些特点：

1. Map函数的输入是通过stand input一行一行的接收数据的。（不像Java API，通过InputFormat类做预处理，使得Map函数的输入是有Key和value的）
2. Map函数的output则必须限定为key-value pair，key和value之间用\t分开。（MapReduce框架在处理intermediate的Map输出时，必须做sort和partition，即shuffle）
3. Reduce函数的input是Map函数的output也是key-value pair，key和value之间用\t分开。
   

常用的Streaming编程语言：

1. bash shell
2. ruby
3. 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 

1. 从标准输入读入一行数据
2. 数据是用\t分隔的键值对
3. 数据是被MapReduce根据key排序之后顺序一行一行读入
4. 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

1. hadoop jar $HADOOP_INSTALL/contrib/streaming/hadoop-*-streaming.jar指明了使用hadoop streaming
2. hadoop-*-streaming.jar会将input里的文件，一行一行的输出到标准输出。
3. 用-mapper指定Map函数。类似于通过管道将数据传给rb文件： data|ch02/src/main/ruby/max_temperature_map.rb
4. -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

1. 这里的-D mapred.reduce.tasks=0将reduce task观掉，因此也不需要设置-reducer
2. 只使用Mapper，可以通过MapReduce帮助我们并行的完成一些平时只能串行的shell脚本
3. 注意这里的-file，在集群模式下，需要并行运行时，需要-file把文件传输到其他节点
   

Combiner

在streaming模式下，仍然可以运行Combiner，两种方法：

1. 通过Java编写一个combiner的函数，并使用-combiner option
2. 以命令行的管道模式完成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

感谢阅读，希望能帮助到大家，谢谢大家对本站的支持！

# Hadoop  # streaming详细介绍  # streaming  # streaming详解  # Hadoop组件简介  # Hadoop-3.1.2完全分布式环境搭建过程图文详解（Windows 10）  # Hadoop源码分析一架构关系简介  # 键值  # 类似于  # 编程语言  # 是一个  # 也不  # 是有  # 模式下  # 两种  # 可以用  # 适用于  # 并在  # 可以通过  # 希望能  # 不像  # 仍是  # 使我  # 种方法  # 发送到  # 谢谢大家  # 会将 

相关文章： 音响网站制作视频教程,隆霸音响官方网站？  如何获取开源自助建站系统免费下载链接？  如何在Ubuntu系统下快速搭建WordPress个人网站？  微信推文制作网站有哪些,怎么做微信推文，急？  如何在Golang中引入测试模块_Golang测试包导入与使用实践  招贴海报怎么做,什么是海报招贴？  成都网站制作价格表,现在成都广电的单独网络宽带有多少的，资费是什么情况呢？  如何在腾讯云服务器上快速搭建个人网站？  建站主机无法访问？如何排查域名与服务器问题  如何在IIS中配置站点IP、端口及主机头？  历史网站制作软件,华为如何找回被删除的网站？  如何通过万网虚拟主机快速搭建网站？  上海网站制作网站建设公司,建筑电工证网上查询系统入口？  品牌网站制作公司有哪些,买正品品牌一般去哪个网站买？  独立制作一个网站多少钱,建立网站需要花多少钱？  建站与域名管理如何高效结合？  大型企业网站制作流程,做网站需要注册公司吗？  如何选择适合PHP云建站的开源框架？  齐河建站公司：营销型网站建设与SEO优化双核驱动策略  如何在阿里云虚拟服务器快速搭建网站？  如何在阿里云虚拟主机上快速搭建个人网站？  如何通过IIS搭建网站并配置访问权限？  建站之星代理商如何保障技术支持与售后服务？  如何选择服务器才能高效搭建专属网站？  如何基于云服务器快速搭建个人网站？  深圳网站制作培训,深圳哪些招聘网站比较好？  如何零基础开发自助建站系统？完整教程解析  ,如何利用word制作宣传手册？  如何快速搭建虚拟主机网站？新手必看指南  常州企业建站如何选择最佳模板？  如何配置WinSCP新建站点的密钥验证步骤？  专业商城网站制作公司有哪些,pi商城官网是哪个？  ,网站推广常用方法？  如何在自有机房高效搭建专业网站？  制作企业网站建设方案,怎样建设一个公司网站？  制作网站的模板软件,网站怎么建设？  制作充值网站的软件,做人力招聘为什么要自己交端口钱？  企业网站制作公司网页,推荐几家专业的天津网站制作公司？  如何选择CMS系统实现快速建站与SEO优化？  如何彻底删除建站之星生成的Banner？  高性能网站服务器配置指南：安全稳定与高效建站核心方案  怎么制作网站设计模板图片,有电商商品详情页面的免费模板素材网站推荐吗？  建站之星在线客服如何快速接入解答？  建站之星安装提示数据库无法连接如何解决？  如何做静态网页,sublimetext3.0制作静态网页？  深入理解Android中的xmlns:tools属性  怎么用手机制作网站链接,dw怎么把手机适应页面变成网页？  如何正确选择百度移动适配建站域名？  微信h5制作网站有哪些,免费微信H5页面制作工具？  武汉网站设计制作公司,武汉有哪些比较大的同城网站或论坛，就是里面都是武汉人的？