Hadoop-Streaming
所有文档

          百度MapReduce BMR

          Hadoop-Streaming

          Hadoop Streaming简介

          本文以分析Web日志统计每日请求量为例,介绍如何在百度智能云平台使用Hadoop Streaming。

          在BMR集群中,您可以使用python、shell、C++等任何您熟悉的编程语言开发Hadoop Streaming作业。Hadoop Streaming是Hadoop提供的编程工具,允许用户使用任何可执行文件或者脚本文件作为Mapper和Reducer,它将Mapper和Reducer文件封装成MapReduce作业并提交运行,让您无须打包即可快速实现MapReduce功能,满足复杂的业务需求。

          程序准备

          Mapper.py程序:

              #!/usr/bin/env python
              import sys
              import re
              separator = "\\s{1}"
              ipAddr = "(\\S+)"
              remoteUser = "(.*?)"
              timeLocal = "\\[(.*?)\\]"
              request = "(.*?)"
              status = "(\\d{3})"
              bodyBytesSent = "(\\S+)"
              httpReferer = "(.*?)"
              httpCookie = "(.*?)"
              httpUserAgent = "(.*?)"
              requestTime = "(\\S+)"
              host = "(\\S+)"
              msec = "(\\S+)"
              p = ipAddr + separator + "-" + separator + timeLocal\
                  + separator + request + separator + status + separator\
                  + bodyBytesSent + separator + httpReferer + separator + httpCookie\
                  + separator + remoteUser + separator + httpUserAgent + separator\
                  + requestTime + separator + host + separator + msec
              for line in sys.stdin:
                  line = line.strip()
                  m = re.match(p, line)
                  if m is not None:
                      print '%s\t%s' % (m.group(2).split(":")[0], 1)

          Reducer.py程序:

              #!/usr/bin/env python
              import sys
              from operator import itemgetter
              sum_result = {}
              for line in sys.stdin:
                  line = line.strip()
                  key, value = line.split()
                  try:
                      value = int(value)
                      sum_result[key] = sum_result.get(key, 0) + value
                  except ValueError:
                      pass
              sorted_sum_result = sorted(sum_result.items(), key=itemgetter(0))
              for key, value in sorted_sum_result:
                  print '%s\t%s'% (key, value)

          集群准备

          1. 准备数据,请参考数据准备
          2. 百度智能云环境准备
          3. 登录控制台,选择“产品服务->百度MapReduce BMR”,点击“创建集群”,进入集群创建页,并做如下配置:

            • 设置集群名称
            • 设置管理员密码
            • 关闭日志开关
            • 选择镜像版本“BMR 1.0.0(hadoop 2.7)”
            • 选择内置模板“hadoop”
          4. 请保持集群的其他默认配置不变,点击“完成”可在集群列表页查看已创建的集群,当集群状态由“初始化中”变为“空闲中”时,集群创建成功。

          运行作业

          1. 在“产品服务>MapReduce>百度MapReduce-作业列表”页中,点击“创建作业”,进入创建作业页。
          2. 配置Streaming作业参数:

            • 作业类型:选择“Streaming作业”;
            • 作业名称:输入作业名称,长度不可超过255个字符;
            • Mapper:若使用您自己的代码,请上传程序至BOS或者您本地的HDFS中,并在此输入程序路径;您也可直接使用百度智能云提供的样例程序,路径如下: 华北-北京区域的BMR集群对应的样例程序路径:bos://bmr-public-bj/sample/streaming-1.0-mapper.py 华南-广州区域的BMR集群对应的样例程序路径:bos://bmr-public-gz/sample/streaming-1.0-mapper.py
            • Reducer:若使用您自己的代码,请上传程序至BOS或者您本地的HDFS中,并在此输入程序路径;您也可直接使用百度智能云提供的样例程序,路径如下: 华北-北京区域的BMR集群对应的样例程序路径:bos://bmr-public-bj/sample/streaming-1.0-reducer.py 华南-广州区域的BMR集群对应的样例程序路径:bos://bmr-public-gz/sample/streaming-1.0-reducer.py
            • BOS输入地址:bos://bmr-public-bj/data/log/accesslog-1k.log
            • BOS输出地址:输出路径必须具有写权限且该路径不能已存在,bos://{your-bucket}/output
            • 失败后操作:继续;
            • 应用程序参数:无。
          3. 在“集群适配”区,选择适配的集群。
          4. 点击“完成”,则作业创建成功;运行中的作业状态会由“等待中”更新为“运行中”,当作业运行完毕后,状态会更新为“已完成”,即可查看到结果了。

          查看结果

          您可以到您所选的存储系统(BOS或HDFS)中查看输出结果,以下是在BOS中查看输出结果的说明:

          用户到bos://{your-bucket}/output路径下查看输出,如果使用系统提供的输入数据和程序,可以打开输出结果看到如下内容:

          03/Oct/2015    139
          04/Oct/2015    375
          05/Oct/2015    372
          06/Oct/2015    114

          分布式缓存导入数据

          分布式缓存是Hadoop提供的文件缓存工具,它能够自动将指定的文件分发到运行Map或Reduce任务的各个节点上,并缓存到本地,供用户程序读取使用。在Map或Reduce时,如需访问通用数据,利用分布式缓存可显著提高访问效率。下面以Streaming作业为例说明如何使用Hadoop分布式缓存。

          应用场景

          分布式缓存通常用在当MapReduce需要依赖一些特定的版本库,或者MapReduce需要访问第三方库或文件时,若将这些库或文件安装到集群各个机器上通常比较麻烦,这时可以使用分布式缓存将其分发到集群各个节点上,程序运行完后,Hadoop自动将其删除。

          例如下述的场景,运用分布式缓存会很好的解决问题:

          • 分发字典文件:Map或者Reduce需要用到一些外部字典时,可使用分发字典文件,比如黑白名单、词表等。
          • 自动化软件部署:MapReduce需依赖于特定版本的库时,可使用自动化软件部署,比如依赖于某个版本的PHP解释器,可以运用分布式缓存进行上传分发。

          支持的文件类型

          分布式缓存支持单个文件和打包文件,支持以下压缩格式:

          * zip
          * tgz
          * tar.gz
          * tar
          * jar

          操作步骤

          1. 您需要先将文件上传到BOS,可参考BOS上传Object
          2. 提交作业时通过指定参数,为缓存文件指定一个软链接,在Mapper和Reducer中通过调用软连接即可访问实际的文件内容;

            • -files:通常用来上传单个文件,可将指定文件分发到各个Task的工作目录下,不做其他处理;如需上传多个文件,文件之间用逗号隔开;
            • -archives:通常用来上传打包文件,可将指定文件分发到各个Task的工作目录下,并对后缀名为“.jar”、“.zip”,“.tar.gz”、“.tgz”的文件自动解压,默认情况下,解压后的内容存放到工作目录下名称为解压前文件名的目录中;如需上传多个打包文件,文件之间用逗号隔开。
          文件类型 操作 说明
          添加单个文件 输入参数:-files+文件在BOS中的位置+井号(#)+文件在缓存中的软链接名称 -files bos://bucket_name/file_name#file_name
          添加打包文件 输入参数:-archives+文件在BOS中的位置+井号(#)+文件在缓存中的软链接名称 -archives bos://bucket_name/archive_name#archive_name

          例如:

          • 对于某一个文件mydata.txt,将其上传到BOS文件系统,路径为bos://mybucket/mydata.txt,提交MapReduce Streaming作业时在参数中指定 -files bos://mybucket/mydata.txt#data,在MapTask和ReduceTask中就可以通过 ./data 来访问所需的文件了;
          • 对于某一个文件libA.so,将其打包到mylib.tar.gz文件中,并上传到BOS文件系统,路径为bos://mybucket/mylib.tar.gz,提交MapReduce Streaming作业时在参数中指定 -archives bos://mybucket/mylib.tar.gz#libs,在MapTask和ReduceTask中就可以通过./libs/libA.so来访问所需的文件了。
          上一篇
          快速入门
          下一篇
          Spark