本文主要讲解三个问题:
      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 = 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\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 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 目录.

最新文章

  1. iOS开发~CocoaPods使用详细说明
  2. 在 Typescript 2.0 中使用 @types 类型定义
  3. 【T-SQL基础】02.联接查询
  4. [QGLViewer]First Demo
  5. POSIX字符类
  6. 安装ORACLE后,改变计算机名称,导致OracleDBConsoleOrcl服务无法启动
  7. JDBC数据更新
  8. dell 交换机 双链路冗余
  9. POJ 1845 Sumdiv (求某个数的所有正因子的和)
  10. CORS 跨域 实现思路及相关解决方案
  11. VC 项目支撑文件解释
  12. 通过YAJL获取json中的值
  13. 基于表单的身份验证(FBA)
  14. iOS8的屏幕旋转的问题
  15. CSS如何实现圆角的outline效果?
  16. 使用oracle DB_LINK的一个注意点
  17. Android开发技巧——使用Drawable实现小红点
  18. 【XSY2887】【GDOI2018】小学生图论题 分治FFT 多项式exp
  19. sql server开发工具
  20. zepto.js

热门文章

  1. 解决mac上matplotlib中文无法显示问题
  2. SSH密钥登陆免密码方法
  3. android之代码混淆
  4. nmcli 使用记录---fatt
  5. vue项目搭建 (二) axios 封装篇
  6. Spring入门5.事务管理机制
  7. SQL SERVER 算法执行效率
  8. gitblit-部署
  9. 使用VisualStudio读写NI FPGA板卡实例(基于FPGA Interface C API Generator)
  10. [置顶] kubernetes1.7新特性:日志审计变化