MapReduce开发总结

2018-06-03

MapReduce开发总结

1.在java实现的MapReduce程序中，需要写一个继承 org.apache.hadoop.mapreduce.Mapper的类和一个继承 org.apache.hadoop.mapreduce.Reducer的类，在实现的类中分别重写map和reduce方法。创建一个含有主函数入口的类，在主函数中进行MapReduce任务相关的配置。
主要配置如下几个条目：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;


public class MyWapResin {

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		//Job job = new Job(conf); //老版本方法
		job.setJarByClass(MyWapResin.class); //设置Jar的入口类
		job.setJobName("MyWapResin"); 
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); //因为在运行jar时有 -Dproperty=value 参数，将其过滤掉

		FileInputFormat.addInputPaths(job, otherArgs[0]); //输入文件的路径，多条路径之间用逗号分隔
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); //输出文件的路径，要求定位在一个原本不存在的路径下，否个运行时会报错
		
		job.setMapperClass(MyWapResinMapper.class);//指定Mppaer类
		job.setReducerClass(MyWapResinReducer.class);//指定Reducer类
		//设置输入key：value的类型，可以省略不设置
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		//设置Reducer输出的key:value的类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		//设置Reducer任务的个数，决定最后输出有多少个文件
		job.setNumReduceTasks(100);
		//等待程序结束，参数true表示打印出进度等相关信息
		boolean res = job.waitForCompletion(true);
		System.exit(res ? 0 : 1);
	}

}

2.MapReduce运行过程的输出信息中有一个条目是：
The url to track the job: http://xxx.xxx.xxx_xxx_xxx/
在浏览器中打开对应的url可以查看该任务的运行相关日志，方便对程序进行debug。在debug过程中，我们可能需要在实现mapreduce的程序中使用 System.out.println("xxx") 来查看程序执行的状态，该输出信息不会被输出到标准输出中，而是输出在hadoop安装的目录 logs/userlogs/ 对应任务的 stdout 文件中，也可指直接在对应的url上查看对应的log信息中查看到。

3.在elipse编译含有main函数的类时，出现 “找不到或无法加载主类” 的错误，此时在左侧栏的该项目文件标志上方有个红色的感叹号。。主要原因是，原来从外部加载的一些jar文件找不到了（外部的jar文件位置发生变化或者被删除）。解决办法是，在该项目下鼠标右键 Properties->Java Build Path->Libraries,将其中丢失的jar包移除即可。

4.将相关的项目使用eclipse导出jar包，需要指定入口的main class。通过命令 hadoop jar MyWapResin.jar [-Dmapreduce.map.memory.mb=4096 -Dmapreduce.reduce.memory.mb=8192 -Dmapreduce.map.java.opts=-Xmx3072m -Dmapreduce.reduce.java.opts=-Xmx6144m -Dmapreduce.min.split.size=3000g ...] inputPath outputPath 执行相应的MapReduce程序。

5.通过Xshell等其他工具登录到开发机上时，此时Linux会开启一个进程（父进程），此后运行的各种程序都是该进程（父进程）的子进程，一旦网络不稳定等各种原因导致本地与开发机断开连接，对应的父进程就会被关闭，因而相关的子进程也就被关闭了，导致某些需要长时间执行的子进程不能正常执行。解决办法：

nohup ./xxx.sh &    #让提交的命令忽略hangup信号，&让进程转到后台运行，输出信息会定位在当前目录下的 nohup.out 文件中
setsid ./xxx.sh  #将该进程的父进程指定为 1 （init进程）
[ctrl+z]  #将当前运行的进程挂起
jobs  #查看当前所有的进程
bg xxxx  #将进程 xxxx 转入后台执行
fg xxxx  #将进程 xxxx 撞到前台执行

6.使用hadoop-streaming.jar运行Python编写的mapreduce程序相关设置：
用两个.py文件分别实现mapper和reducer对应的操作，其输入都来自于标准输入 stdin 。与java不同的是，在reduce阶段，java的输入是 {key_1, list(value_1, value_2, …)}, 此时reducer获得的values是一堆value的集合；而Python中reducer阶段输入的是{key_1, value_1}，就是mapper的输出一行数据，只不过在mapper在很多行数据传给reducer之前会根据key进行排序，然后依次喂给reducer，所以reducer处理的数据中key相同的数据会处于相邻的行。
可以通过在放到集群上运行之前在小数据上通过 cat data.txt | ./mapper.py | sort | ./reducer.py > output.txt 进行测试。
shell 脚本配置示例：

stream="/usr/lib/hadoop-mapreduce/hadoop-streaming.jar" #根据hadoop实际安装的位置进行配置
inputPath="xxx/xxx/" #根据实际情况配置
outputPath="xxxx/xxx/"

hadoop jar $stream \
-D mapreduce.job.reduces=300 \
-D mapreduce.input.fileinputformat.split.minsize=134217728 \
-D mapreduce.input.fileinputformat.split.maxsize=134217728 \
-D mapreduce.job.name="MyWapResin_step1" \
-D mapreduce.output.fileoutputformat.compress=true \
-D mapreduce.output.fileoutputformat.compress.codec=com.hadoop.compression.lzo.LzopCodec \ #设置输出文件以lzo格式进行压缩
-mapper MyWapResinMapper_step1.py \
-reducer MyWapResinReducer_step1.py  \
-input $inputFile \
-output $outputFile \
-file MyWapResinMapper_step1.py \
-file MyWapResinReducer_step1.py

Tips：
如果命令行过长的话，可以先使用字符串表示出，例：
cmd_string=”cat data.txt | head -n 10”
eval $cmd_string #通过eval使得cmd_string生效