zoukankan      html  css  js  c++  java
  • 如何向map和reduce脚本传递参数,加载文件和目录

    本文主要讲解三个问题:
          1 使用Java编写MapReduce程序时,如何向map、reduce函数传递参数。
          2 使用Streaming编写MapReduce程序(C/C++, Shell, Python)时,如何向map、reduce脚本传递参数
          3 使用Streaming编写MapReduce程序(C/C++, Shell, Python)时,如何向map、reduce脚本传递文件或文件夹
             (1) streaming 加载本地单个文件
             (2) streaming 加载本地多个文件
             (3) streaming 加载本地目录
             (4) streaming编程时在mapreduce脚本中读 hdfs 文件
             (5) streaming编程时在mapreduce脚本中读 hdfs 目录
     
    1.  Java编写MapReduce程序时,如何向map、reduce函数传递参数

    我开始使用如下方式进行传递.

    在主类中声明两个静态变量, 然后在 main 函数中给变量赋值, 试图在 map和reduce函数中获得变量的值。
    代码结构类似如下:
    [MapReduce] <wbr>如何向map和reduce脚本传递参数,加载文件和目录《转载》
     
     
     
     
    提交到集群运行发现在 map 和 reduce函数中, 静态变量MaxScore的值始终是初值1。
    于是试图在主类的静态区中给变量赋值 (因为静态区中的代码比main中的代码要先执行), 仍是不成功, MaxScore的值始终是初值1。
    将上述代码在 单机hadoop上运行, 结果正常, map 函数中能获得变量的值。
    思考是这个原因: 在提交作业到hadoop集群后,mapper类和reducer类就到各个 tasktracker上去运行了, 与主类独立, 不能交互。
    因此,上述往 map 和 reduce 函数传参数的方法实在太天真。
    于是想到其它一些方法: 例如将参数写入hdfs文件中, 然后在 mapper 和 reducer 类的 run方法中读取文件, 并将值读到相应变量,这是可行的,但是方法较复杂,代码如下:
    上述方法尽管可用, 但是不是常规方法, 下面介绍常用的方法:
    (1) 通过 Configuration 来传递参数
    在main函数中调用set方法设置参数, 例如:

    在mapper中通过上下文context来获取当前作业的配置, 并获取参数, 例如:

    注: context 很有用, 能获取当前作业的大量信息,例如上面就获取了任务ID.
     
    (2) 利用org.apache.hadoop.io.DefaultStringifier类

    示例:

    main中:

        Configuration conf = new Configuration();

        Text maxscore = new Text("12989");

        DefaultStringifier.store(conf, maxscore ,"maxscore");

    这样,Text对象maxscore就以“maxscore”作为key存储在conf对象中了,然后在map和reduce函数中调用load的方法便可以把对象读出。

    mapper获取:

        Configuration conf = context.getConfiguration()

        Text out = DefaultStringifier.load(conf, "maxscore", Text.class);

       需要说明的是,这个需要传递的对象必须要先实现序列化的接口,Hadoop的序列化是通过Writable接口来实现的

    (2) 参考自:http://blog.sina.com.cn/s/blog_6b7cf18f0100x9jg.html

     

    2.  编写 Streaming 程序时,如何向map、reduce函数传递参数

    可以通过 streaming 的 cmdenv 选项设置环境变量,然后在 map 和 reduce 脚本中获取环境变量。

    可参考 << hadoop streaming 高级编程 >>

    http://dongxicheng.org/mapreduce/hadoop-streaming-advanced-programming/

    (0) 作业提交脚本:

    #!/usr/bin/env bash

    max_read_count=${array[0]}

    min_read_count=${array[1]}

    max_write_count=${array[2]}

    min_write_count=${array[3]}

    hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-0.20.2-streaming.jar

      -D mapred.reduce.tasks=1   

      -input $input

      -output $output  

      -mapper $mapper_script

      -file $map_file  

      -reducer $reducer_script

      -file $reduce_file

      -cmdenv "max_read_count=${array[0]}"      # 设置环境变量   max_read_count .   

      -cmdenv "min_read_count=${array[1]}"        # 多个变量时请多次使用 -cmdenv   

      -cmdenv "max_write_count=${array[2]}"

      -cmdenv "min_write_count=${array[3]}"  

    (1) Python mapper.py

    #!/usr/bin/env python

    import sys

    import os

    min_r_count = float(os.environ.get('min_read_count')) # get environment variables.

    max_r_count = float(os.environ.get('max_read_count'))

    min_w_count = float(os.environ.get('min_write_count'))

    max_w_count = float(os.environ.get('max_write_count'))

    (2)Shell mapper.sh

    #!/usr/bin/env bash 

    while read line  # 读入行

    do

       a=$line  

    done

    echo $min_read_count $max_read_count  # get environment variables.

    (3)C/C++ mapper.c

    #include

    #include

    int main(int argc, char *argv[], char *env[])

    {

      double min_r_count;

      int i = 0;

      for (i = 0; env[i] != NULL; i++) // env[i] 存储了环境变量, 每项的值为此种形式: PATH=******, 所以需要截取变量值

      { 

         if( strstr(env[i], "PATH=") ) {

           char *p =NULL;

           p = strstr(env[i], "=");

           if( (p-env[i]) == 4 )

             printf("%s ", ++p); // 获取 PATH 环境变量

         }

          if( strstr(env[i], "min_write_count=") ) {

           char *p =NULL;

           p = strstr(env[i], "=");

           if( (p-env[i]) == strlen("min_write_count") )

             printf("%s ", ++p); // 获取  min_write_count  环境变量

         }

      }

      char eachLine[200]={0};

      while(fgets(eachLine, 199, stdin)) // read line from stdin

      {

         printf("%s", eachLine);

      }

    }


    注意:
        Hadoop执行命令时的选项有顺序的, 顺序是 bin/hadoop command [genericOptions] [commandOptions]. 
        对于streaming, -D 属于genericOptions, 即hadoop的通用选项, 所以必须写在前面.
        Streaming 的所有选项可以参考:  
        hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-0.20.2-streaming.jar -info
     
    3.  编写 Streaming 程序时,如何向map、reduce函数传递文件或文件夹。

    (1) streaming 加载本地单个文件

    streaming 支持 -file 选项, 可以把 -file 后面的本地文件(注意是本地文件)打包成作业提交的一部分, 即打包到作业的jar文件当中, 这样在mapreduce脚本中就可以像访问本地文件一样访问打包的文件了.

    实例:

    作业提交文件 run.sh

     

    mapper.py

    注意:在提交作业时使用的是 -file logs/wbscoretest.log 指定需要加载的文件. 在 map 脚本中只需要直接读取文件 wbscoretest.log 即可, 不需要写 logs/wbscoretest.log, 因为只加载了文件 wbscoretest.log, 而不会加载 logs 目录和 

    wbscoretest.log 文件.

    (2) streaming 加载本地多个文件

    (3) streaming 加载本地目录 ( 若加载多个目录,用逗号隔开,-files dir1, dir2, dir3 )

    使用streaming的 -file 选项不能加载本地目录, 我实验是如此.

    我们可以使用 hadoop 的通用选项 -files 来加载本地目录, 加载成功后在mapreduce脚本中可以像访问本地目录一样访问加载的目录.

    实际应用中,我们在编写 分词MapReduce作业时需要加载分词词典,就使用该方法.

    作业提交脚本:  

    map 脚本: 读取目录下的文件.

    加载多个目录:

    注意:多个目录之间用逗号隔开,且不能有空格,否则会出错,这个限制太蛋疼了。

    例如:

    (4) streaming编程时在mapreduce脚本中读 hdfs 文件

    使用 -files 选项, 后面跟需要读的 hdfs 文件路径. 这样在 mapreduce 脚本中就可以直接通过文件名来访问该文件.

    作业提交脚本:

    map脚本: 

    如果需要加载大文件, 我们可以将文件先上传到 hdfs 中, 然后在 mapreduce 脚本中读取 hdfs 文件.

    (5) streaming编程时在mapreduce脚本中读 hdfs 目录

    使用 -files 选项, 后面跟需要读的 hdfs 目录. 这样在 mapreduce 脚本中就可以像访问本地目录一样访问该目录.

    作业提交脚本:

    map脚本:  直接读取 tmp_kentzhan 目录.

  • 相关阅读:
    Codeforces 543E. Listening to Music
    UOJ #138. 【UER #3】开学前的涂鸦
    bzoj 3569: DZY Loves Chinese II
    bzoj 2428: [HAOI2006]均分数据
    bzoj 4589: Hard Nim
    UOJ #119. 【UR #8】决战圆锥曲线
    spoj5973
    codeforces555E
    poj1275
    bzoj4152
  • 原文地址:https://www.cnblogs.com/snowbook/p/5762152.html
Copyright © 2011-2022 走看看