Hadoop Streaming框架使用

简介:

Streaming简介 

Streaming框架允许任何程序语言实现的程序在Hadoop MapReduce中使用,方便已有程序向Hadoop平台移植。因此可以说对于hadoop的扩展性意义重大,今天简单说一下。

Streaming的原理是用Java实现一个包装用户程序的MapReduce程序,该程序负责调用MapReduce Java接口获取key/value对输入,创建一个新的进程启动包装的用户程序,将数据通过管道传递给包装的用户程序处理,然后调用MapReduce Java接口将用户程序的输出切分成key/value对输出。 

 

Streaming优点

1 开发效率高,便于移植

只要按照标准输入输出格式进行编程,就可以满足hadoop要求。因此单机程序稍加改动就可以在集群上进行使用。 同样便于测试

只要按照 cat input | mapper | sort | reducer > output 进行单机测试即可。

如果单机测试通过,大多数情况是可以在集群上成功运行的,只要控制好内存就好了。

    2 提高程序效率

有些程序对内存要求较高,如果用java控制内存毕竟不如C/C++。

Streaming不足

    1 Hadoop Streaming默认只能处理文本数据,无法直接对二进制数据进行处理 

    2 Streaming中的mapperreducer默认只能向标准输出写数据,不能方便地处理多路输出 

具体参数介绍

 

-input    <path>

输入数据路径

-output   <path>

输出数据路径

-mapper  <cmd|JavaClassName>

mapper可执行程序或Java类

-reducer  <cmd|JavaClassName>

reducer可执行程序或Java类

-file            <file>        Optional

分发本地文件

-cacheFile       <file>       Optional

分发HDFS文件

-cacheArchive   <file>        Optional

分发HDFS压缩文件

-numReduceTasks  <num>    Optional

reduce任务个数

-jobconf | -D NAME=VALUE    Optional

作业配置参数

-combiner <JavaClassName>   Optional

Combiner Java

-partitioner <JavaClassName>  Optional

Partitioner Java

-inputformat <JavaClassName> Optional

InputFormat Java

-outputformat <JavaClassName>Optional

OutputFormat Java

-inputreader <spec>            Optional

InputReader配置

-cmdenv   <n>=<v>           Optional

传给mapper和reducer的环境变量

-mapdebug <path>             Optional

mapper失败时运行的debug程序

-reducedebug <path>           Optional

reducer失败时运行的debug程序

-verbose                      Optional

详细输出模式

 

 下面是对各个参数的详细说明:

-input <path>:指定作业输入,path可以是文件或者目录,可以使用*通配符,-input选项可以使用多次指定多个文件或目录作为输入。

-output <path>:指定作业输出目录,path必须不存在,而且执行作业的用户必须有创建该目录的权限,-output只能使用一次。

-mapper:指定mapper可执行程序或Java类,必须指定且唯一。

-reducer:指定reducer可执行程序或Java类,必须指定且唯一。

-file, -cacheFile, -cacheArchive:分别用于向计算节点分发本地文件、HDFS文件和HDFS压缩文件

-numReduceTasks:指定reducer的个数,如果设置-numReduceTasks 0或者-reducer NONE则没有reducer程序,mapper的输出直接作为整个作业的输出。

-jobconf | -D NAME=VALUE:指定作业参数,NAME是参数名,VALUE是参数值,可以指定的参数参考hadoop-default.xml特别建议-jobconf mapred.job.name='My Job Name'设置作业名,使用-jobconf mapred.job.priority=VERY_HIGH | HIGH | NORMAL | LOW | VERY_LOW设置作业优先级,使用-jobconf mapred.job.map.capacity=M设置同时最多运行Mmap任务,使用-jobconf mapred.job.reduce.capacity=N设置同时最多运行Nreduce任务。

常见的作业配置参数如下表所示: 

mapred.job.name

作业名

mapred.job.priority

作业优先级

mapred.job.map.capacity

最多同时运行map任务数

mapred.job.reduce.capacity

最多同时运行reduce任务数

hadoop.job.ugi

作业执行权限

mapred.map.tasks

map任务个数

mapred.reduce.tasks

reduce任务个数

mapred.job.groups

作业可运行的计算节点分组

mapred.task.timeout

任务没有响应(输入输出)的最大时间

mapred.compress.map.output

map的输出是否压缩

mapred.map.output.compression.codec

map的输出压缩方式

mapred.output.compress

reduce的输出是否压缩

mapred.output.compression.codec

reduce的输出压缩方式

stream.map.output.field.separator

map输出分隔符

 -combiner:指定combiner Java类,对应的Java类文件打包成jar文件后用-file分发。

-partitioner:指定partitioner Java类,Streaming提供了一些实用的partitioner实现,参考KeyBasedFiledPartitonerIntHashPartitioner

-inputformat, -outputformat:指定inputformatoutputformat Java类,用于读取输入数据和写入输出数据,分别要实现InputFormatOutputFormat接口。如果不指定,默认使用TextInputFormatTextOutputFormat

-cmdenv NAME=VALUE:给mapperreducer程序传递额外的环境变量,NAME是变量名,VALUE是变量值。

-mapdebug, -reducedebug:分别指定mapperreducer程序失败时运行的debug程序。

-verbose:指定输出详细信息,例如分发哪些文件,实际作业配置参数值等,可以用于调试。




提交hadoop任务示例: 

  $HADOOP_HOME/bin/hadoop streaming \

-input /user/test/input -output /user/test/output \ 

-mapper “mymapper.sh” -reducer “myreducer.sh” \

-file/home/work/mymapper.sh \

-file /home/work/myreducer.sh \

-jobconf mapred.job.name=”file-demo” 

上面的命令提交了一个hadoop任务,输出和输入分别为 /user/test/output 和/user/test/input。 map程序为 mymapper.sh,reduce程序为myreducer.sh。这里需要注意一定要将这两个文件用-file分发到集群的节点上。最后一行指定了任务的名字。

 

还有一些较为复杂的使用,比如需要指定任务个数等,可以使用

-jobconf mapred.job.map.capacity=m -jobconf mapred.job.reduce.capacity=n 

上面的命令设置最多同时运行mmap任务,nreduce任务,如果mn0或者没有指定,则对应的capacity没有限制,默认配置就是0没有限制。建议在运行作业时都设置mapreduce capacity,防止作业占用过多资源。 

 





前两篇文章介绍了Hadoop Streaming框架的使用方法。由于篇幅所限,并没有介绍其中的高级使用方法,但是有一些用法还是相当常见的。今天对一些高级用法进行一个简单的说明,希望能给大家一些启发。

  1 使用cacheFile分发文件

  如果文件(如字典文件)存放在HDFS中,希望计算时在每个计算节点上将文件当作本地文件处理,,可以使用-cacheFile hdfs://host:port/path/to/file#linkname选项在计算节点缓存文件,Streaming程序通过./linkname访问文件。

  例如:

   hadoop = `which hadoop`

   $hadoop streaming \

   -input /user/test/input -output /user/test/output \

-mapper mymapper.sh -reducer myreducer.sh \

   -file /home/work/mymapper.sh \ 

-file /home/work/myreducer.sh \

-cacheFile hdfs://namenode:port/user/test/dict.data#dictlink \

-jobconf mapred.job.name=”cache-file-demo”

  mymapper.shmyreducer.sh可以通过./dictlink直接访问字典文件hdfs://user/test/dict.data,而且是从本地读取文件。


  2 用cacheArchive分发压缩包

  有时要分发的文件有一定的目录结构,可以先将整个目录打包,然后整体进行上传。使用-cacheArchive hdfs://host:port/path/to/archivefile#linkname分发压缩包。

例如在本地有一个目录为app,里面有mapper.pl, reducer.pl, dict/dict.txt这些子目录和文件,mapper.plreducer.pl要读取./dict/dict.txt文件,希望在任务执行时不需要修改程序和目录结构, 可以按照下面的方式分发app目录:

   $ tar app.tar.gz –C app .  #本地打包

   $ $HADOOP_HOME/bin/hadoop fs –put app.tar.gz /user/test/app.tar.gz   #包上传到HDFS

$ $HADOOP_HOME/bin/hadoop streaming \

-input /user/test/input -output /user/test/output \

-mapper “perl app/mapper.pl” -reducer “perl app/reducer.pl” \

-cacheArchive hdfs://namenode:port/user/test/ app.tar.gz #app \

-jobconf mapred.job.name=”cache-archive-demo”

首先将本地app目录中的所有文件和目录打包压缩,然后上传到HDFS/user/test/app.tar.gz,启动streaming任务时使用-cacheArchive选项将app.tar.gz分发到计算节点并解压到app目录,然后在当前工作目录创建到app目录的链接,-mapper选项指定app/mapper.plmapper程序,-reducer选项指定app/reducer.plreducer程序,它们都可以读取./dict/dict.txt文件。本地打包时要进入目录app而不是在app的上层目录打包,否则要通过app/app/mapper.pl才能访问到mapper.pl文件。

hadoop支持zip, jar, tar.gz格式的压缩包,由于Java解压zip压缩包时会丢失文件权限信息而且遇到中文文件名会出错,所见建议采用tar.gz压缩包。

三种文件分发方式的区别:-file将客户端本地文件打成jar包上传到HDFS然后分发到计算节点,-cacheFileHDFS文件分发到计算节点,-cacheArchiveHDFS压缩文件分发到计算节点并解压。


3输出数据分割

默认情况下Streaming框架将map输出的每一行第一个”\t”之前的部分作为key,之后的部分作为valuekey\tvalue又作为reduce的输入。可以用-D stream.map.output.field.separator改变map输出中keyvalue的分隔符,用-D stream.num.map.output.key.fields设置分隔符的位置,该位置之前的部分作为key,之后的部分作为value。如下所示,其中-D stream.map. output.field.separator=:指定使用冒号”:”map输出的一行分隔为key/value-D stream.num.map.output.key.fields=2指定在第二个冒号处进行分隔,也就是第二个冒号之前的作为key,之后的作为value。如果没有冒号或冒号少于两个,则key为整行,value为空。 

   $HADOOP_HOME/bin/hadoop streaming \

       -D stream.map.output.field.separator=: \

       -D stream.num.map.output.key.fields=2 \

-input /user/test/input -output /user/test/output \

-mapper mymapper.sh -reducer myreducer.sh \

-file /home/work/mymapper.sh \

-file /home/work/myreducer.sh \

-jobconf mapred.job.name=”output-sep-demo”

map类似,对于reduce的输出,同样也可以用-D stream.reduce.output.field.separator-D stream.num.reduce.output.key.fields定制key/value分隔方式。


二次排序

  KeyFieldBasedPartitionerHadoop库中的一个实用Partitioner,配置相应的参数就可以使用,通过KeyFieldBasedPartitioner可以方便地实现二次排序。 

   $HADOOP_HOME/bin/hadoop streaming \

       -D stream.map.output.field.separator=. \

       -D stream.num.map.output.key.fields=4 \

       -D map.output.key.field.separator=. \

       -D num.key.fields.for.partition=2 \

   -input /user/test/input -output /user/test/output \

   -mapper “mymapper.sh” -reducer “ myreducer.sh” \

-partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner \

-file /home/work/mymapper.sh \

-file /home/work/myreducer.sh \

-jobconf mapred.job.name=”key-partition-demo”

  其中-Dstream.map.output.field.separator=.-D stream.num.map.output.key.fields=4与上面的定制输出数据分隔方式意义相同,指定map的输出行第4个英文句号”.”之前为key,后面为value-partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner指定使用KeyFieldBasedPartitioner-D map.output.key.field.separator=.指定key的内部用英文句号”.”分隔,-D num.key.fields.for.partition=2指定将key分隔出来的前两个部分而不是整个key用于Partitionerpartition

  以上就是我个人认为hadoop streaming中比较常用的技巧,希望对大家有所帮助,同时也多多补充。


本文转自  陈小龙哈   51CTO博客,原文链接:http://blog.51cto.com/chenxiaolong/1844963
相关文章
|
存储 分布式计算 监控
Hadoop【基础知识 01+02】【分布式文件系统HDFS设计原理+特点+存储原理】(部分图片来源于网络)【分布式计算框架MapReduce核心概念+编程模型+combiner&partitioner+词频统计案例解析与进阶+作业的生命周期】(图片来源于网络)
【4月更文挑战第3天】【分布式文件系统HDFS设计原理+特点+存储原理】(部分图片来源于网络)【分布式计算框架MapReduce核心概念+编程模型+combiner&partitioner+词频统计案例解析与进阶+作业的生命周期】(图片来源于网络)
853 2
|
存储 分布式计算 Hadoop
【大数据处理框架】Hadoop大数据处理框架,包括其底层原理、架构、编程模型、生态圈
【大数据处理框架】Hadoop大数据处理框架,包括其底层原理、架构、编程模型、生态圈
962 0
|
分布式计算 安全 Hadoop
聊聊 hadoop 与 sasl 安全框架
聊聊 hadoop 与 sasl 安全框架
|
分布式计算 Hadoop
Hadoop-27 ZooKeeper集群 集群配置启动 3台云服务器 myid集群 zoo.cfg多节点配置 分布式协调框架 Leader Follower Observer
Hadoop-27 ZooKeeper集群 集群配置启动 3台云服务器 myid集群 zoo.cfg多节点配置 分布式协调框架 Leader Follower Observer
345 1
|
存储 SQL 消息中间件
Hadoop-26 ZooKeeper集群 3台云服务器 基础概念简介与环境的配置使用 架构组成 分布式协调框架 Leader Follower Observer
Hadoop-26 ZooKeeper集群 3台云服务器 基础概念简介与环境的配置使用 架构组成 分布式协调框架 Leader Follower Observer
315 0
|
XML 分布式计算 Hadoop
分布式系统详解--框架(Hadoop-单机版搭建)
分布式系统详解--框架(Hadoop-单机版搭建)
252 0
分布式系统详解--框架(Hadoop-单机版搭建)
|
资源调度 分布式计算 监控
【揭秘Hadoop YARN背后的奥秘!】从零开始,带你深入了解YARN资源管理框架的核心架构与实战应用!
【8月更文挑战第24天】Hadoop YARN(Yet Another Resource Negotiator)是Hadoop生态系统中的资源管理器,为Hadoop集群上的应用提供统一的资源管理和调度框架。YARN通过ResourceManager、NodeManager和ApplicationMaster三大核心组件实现高效集群资源利用及多框架支持。本文剖析YARN架构及组件工作原理,并通过示例代码展示如何运行简单的MapReduce任务,帮助读者深入了解YARN机制及其在大数据处理中的应用价值。
447 0
|
分布式计算 Hadoop 大数据
分布式计算框架比较:Hadoop、Spark 与 Flink
【5月更文挑战第31天】Hadoop是大数据处理的开创性框架,专注于大规模批量数据处理,具有高扩展性和容错性。然而,它在实时任务上表现不足。以下是一个简单的Hadoop MapReduce的WordCount程序示例,展示如何统计文本中单词出现次数。
897 0
|
分布式计算 Hadoop 大数据
大数据处理框架在零售业的应用:Apache Hadoop与Apache Spark
【8月更文挑战第20天】Apache Hadoop和Apache Spark为处理海量零售户数据提供了强大的支持
440 0
|
分布式计算 并行计算 搜索推荐
Hadoop MapReduce计算框架
【5月更文挑战第10天】HadoopMapReduce计算框架
352 3

相关实验场景

更多