打开APP
userphoto
未登录

开通VIP,畅享免费电子书等14项超值服

开通VIP
[MapReduce]如何向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函数中获得变量的值。
代码结构类似如下:
提交到集群运行发现在 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 = newConfiguration();

    Text maxscore = newText("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) Pythonmapper.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)Shellmapper.sh

 

#!/usr/bin/env bash 

while read line  # 读入行

do

   a=$line 

done

echo $min_read_count $max_read_count get environmentvariables.


(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\n", ++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\n", ++p); //获取 min_write_count  环境变量

    }

 

  }

  char eachLine[200]={0};

  while(fgets(eachLine, 199, stdin)) // read linefrom stdin

  {

    printf("%s", eachLine);

  }

}

 

 


注意:
    Hadoop执行命令时的选项有顺序的,顺序是 bin/hadoopcommand [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 目录.

本站仅提供存储服务,所有内容均由用户发布,如发现有害或侵权内容,请点击举报
打开APP,阅读全文并永久保存 查看更多类似文章
猜你喜欢
类似文章
【热】打开小程序,算一算2024你的财运
Darwin Streaming Server for Windows 安装
Hadoop MapReduce开发最佳实践(上篇)
Hadoop集群(第7期)_Eclipse开发环境设置
Eclipse下使用Hadoop单机模式调试MapReduce程序
基于Linux的视频点播系统的实现
Hadoop技术原理总结
更多类似文章 >>
生活服务
热点新闻
分享 收藏 导长图 关注 下载文章
绑定账号成功
后续可登录账号畅享VIP特权!
如果VIP功能使用有故障,
可点击这里联系客服!

联系客服