百度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