ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > Hadoop > 开发一个mapreduce程序

开发一个mapreduce程序

原创 Hadoop 作者：paololiu 时间：2015-08-03 11:34:46 0 删除 编辑

整个项目以hadoop自带的hadoop-mapreduce-examples-2.6.0.jar为例。

2.1新建一个项目

说明：这里没有使用hadoop-eclipse-plugin插件，而是直接新建一个java project，和hadoop-mapreduce-examples-2.6.0.jar一样，完成后打包成jar文件在hadoop集群中运行。

打开eclipse，file---new---java project



2.2新建一个lib文件夹，并将hadoop三个jar复制进去



三个jar包所在hadoop的路径：

hadoop/share/hadoop/yarn/lib/commons-cli-1.2.jar

hadoop/share/hadoop/common/hadoop-common-2.6.0.jar

hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-core-2.6.0.jar


2.3将三个jar包add to build path



2.2jar包入口class

最终我们会将项目打包成一个jar文件放到hadoop集群中去运行。但是在打包jar文件的时候只能指定一个main class，那么每一个需求就要打包成一个jar文件，在使用管理上都很麻烦。通过模仿hadoop自带的样例hadoop-mapreduce-examples-2.6.0.jar，我们可以为jar包创建一个统一的man class，再通过它去调用其他的man class。

点击(此处)折叠或打开

  1. package exhadp;

  3. import org.apache.hadoop.util.ProgramDriver;

  5. public class hadpdriver {

  7.     public static void main(String[] args) {
  8.         // TODO Auto-generated method stub

  10.         int exitCode=-1;
  11.         ProgramDriver pdg=new ProgramDriver();
  12.         
  13.         try{
  14.             pdg.addClass("getcount", getcount.class, "getcount");
  15.             pdg.addClass("getmax", getmax.class, "getmax");
  16.             pdg.addClass("getsum", getsum.class, "getsum");
  17.             exitCode=pdg.run(args);
  18.         }catch (Throwable e){
  19.             e.printStackTrace();
  20.         }
  21.         
  22.         System.exit(exitCode);
  23.     }
  24. }


调用的方式如下：

[grid@dehadp01 ~]$ hadoop jar exhadp.jar getsum TEST_OBJECTS/part-m-00000 getsum_out


2.3案例文件说明

1,SYS,ICOL$,null,20,2,TABLE,2011-09-17 09:46:13.0,2011-09-17 09:58:00.0,2011-09-17:09:46:13,VALID,N,N,N,1,null

2,SYS,I_USER1,null,46,46,INDEX,2011-09-17 09:46:13.0,2011-09-17 09:46:13.0,2011-09-17:09:46:13,VALID,N,N,N,4,null

3,SYS,CON$,null,28,28,TABLE,2011-09-17 09:46:13.0,2011-09-17 10:04:20.0,2011-09-17:09:46:13,VALID,N,N,N,1,null

4,SYS,UNDO$,null,15,15,TABLE,2011-09-17 09:46:13.0,2011-09-17 09:46:13.0,2011-09-17:09:46:13,VALID,N,N,N,1,null

5,SYS,C_COBJ#,null,29,29,CLUSTER,2011-09-17 09:46:13.0,2011-09-17 09:46:13.0,2011-09-17:09:46:13,VALID,N,N,N,5,null


数据源于oracle数据库内的：select rownum id,a.* from dba_objects a

每一行即是一条记录，中间以“,”分隔，第一列表示id，第二列表示owner，第七列表示object_type，倒数第二列表示status。


2.4求每个owner的最大的id，即select owner,max(id) from t group by owner

点击(此处)折叠或打开

  1. package exhadp;

  3. //完全模仿wordcount的样式

  5. import java.io.IOException;

  7. import org.apache.hadoop.conf.Configuration;
  8. import org.apache.hadoop.fs.Path;
  9. import org.apache.hadoop.io.IntWritable;
  10. import org.apache.hadoop.io.Text;
  11. import org.apache.hadoop.mapreduce.Job;
  12. import org.apache.hadoop.mapreduce.Mapper;
  13. import org.apache.hadoop.mapreduce.Reducer;
  14. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  15. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  16. import org.apache.hadoop.util.GenericOptionsParser;

  18. public class getmax {

  20.     // map的过程
  21.     public static class TokenizerMapper extends
  22.             Mapper<Object, Text, Text, IntWritable> {
  23.         public void map(Object key, Text value,
  24.                 Mapper<Object, Text, Text, IntWritable>.Context context)
  25.                 throws IOException, InterruptedException {
  26.             // 将源文件中一行的记录以字符串的形式存入line中
  27.             String line = value.toString();
  28.             String[] strarray = line.split(",");
  29.             // 以“,”为分隔符，第一列（strarray[0]）为id,第二列（strarray[1]）为owner
  30.             Text owner = new Text(strarray[1]);
  31.             int id = Integer.parseInt(strarray[0]);
  32.             // 将owner做为map的key,id作为value
  33.             context.write(owner, new IntWritable(id));
  34.         }
  35.     }

  37.     // reduce过程
  38.     public static class IntMaxReducer extends
  39.             Reducer<Text, IntWritable, Text, IntWritable> {
  40.         public void reduce(Text key, Iterable<IntWritable> values,
  41.                 Reducer<Text, IntWritable, Text, IntWritable>.Context context)
  42.                 throws IOException, InterruptedException {
  43.             int max = Integer.MIN_VALUE;
  44.             for (IntWritable value : values) {
  45.                 max = Math.max(max, value.get());
  46.             }
  47.             context.write(key, new IntWritable(max));
  48.         }
  49.     }

  51.     // 主程序分别调用map和reduce的过程
  52.     public static void main(String[] args) throws Exception {
  53.         Configuration conf = new Configuration();
  54.         String[] otherArgs = new GenericOptionsParser(conf, args)
  55.                 .getRemainingArgs();
  56.         if (otherArgs.length < 2) {
  57.             System.err.println("Usage: getmax <in> [<in>...] <out>");
  58.             System.exit(2);
  59.         }
  60.         Job job = new Job(conf, "getmax");
  61.         job.setJarByClass(getmax.class);
  62.         job.setMapperClass(TokenizerMapper.class);
  63.         job.setCombinerClass(IntMaxReducer.class);
  64.         job.setReducerClass(IntMaxReducer.class);
  65.         job.setOutputKeyClass(Text.class);
  66.         job.setOutputValueClass(IntWritable.class);
  67.         for (int i = 0; i < otherArgs.length - 1; i++) {
  68.             FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
  69.         }
  70.         FileOutputFormat.setOutputPath(job, new Path(
  71.                 otherArgs[(otherArgs.length - 1)]));

  73.         System.exit(job.waitForCompletion(true) ? 0 : 1);
  74.     }
  75. }


2.5统计每种status的数量，即select status,count(*) from t group by status

点击(此处)折叠或打开

  1. package exhadp;

  3. import java.io.IOException;

  5. import org.apache.hadoop.conf.Configuration;
  6. import org.apache.hadoop.fs.Path;
  7. import org.apache.hadoop.io.IntWritable;
  8. import org.apache.hadoop.io.Text;
  9. import org.apache.hadoop.mapreduce.Job;
  10. import org.apache.hadoop.mapreduce.Mapper;
  11. import org.apache.hadoop.mapreduce.Reducer;
  12. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  13. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  14. import org.apache.hadoop.util.GenericOptionsParser;

  16. public class getcount {

  18.     public static void main(String[] args) throws Exception {
  19.         Configuration conf = new Configuration();
  20.         String[] otherArgs = new GenericOptionsParser(conf, args)
  21.                 .getRemainingArgs();
  22.         if (otherArgs.length < 2) {
  23.             System.err.println("Usage: getcount <in> [<in>...] <out>");
  24.             System.exit(2);
  25.         }
  26.         Job job = new Job(conf, "getcount");
  27.         job.setJarByClass(getcount.class);
  28.         job.setMapperClass(TokenizerMapper.class);
  29.         job.setCombinerClass(IntCountReducer.class);
  30.         job.setReducerClass(IntCountReducer.class);
  31.         job.setOutputKeyClass(Text.class);
  32.         job.setOutputValueClass(IntWritable.class);
  33.         for (int i = 0; i < otherArgs.length - 1; i++) {
  34.             FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
  35.         }
  36.         FileOutputFormat.setOutputPath(job, new Path(
  37.                 otherArgs[(otherArgs.length - 1)]));

  39.         System.exit(job.waitForCompletion(true) ? 0 : 1);
  40.     }

  42.     public static class IntCountReducer extends
  43.             Reducer<Text, IntWritable, Text, IntWritable> {
  44.         public void reduce(Text key, Iterable<IntWritable> values,
  45.                 Reducer<Text, IntWritable, Text, IntWritable>.Context context)
  46.                 throws IOException, InterruptedException {
  47.             int count = 0;
  48.             for (IntWritable value : values) {
  49.                 // 这里count累加value的值（就是1）,这样就可以统计出总数出来了
  50.                 count = count + value.get();
  51.             }
  52.             context.write(key, new IntWritable(count));
  53.         }
  54.     }

  56.     public static class TokenizerMapper extends
  57.             Mapper<Object, Text, Text, IntWritable> {
  58.         public void map(Object key, Text value,
  59.                 Mapper<Object, Text, Text, IntWritable>.Context context)
  60.                 throws IOException, InterruptedException {
  61.             String line = value.toString();
  62.             String[] strarray = line.split(",");
  63.             Text status = new Text(strarray[10]);
  64.             // 取第11列的status的值作为map的key,因为只是统计count，所以使用一个常量1作为map的value就可以了
  65.             context.write(status, new IntWritable(1));
  66.         }
  67.     }

  69. }


2.6求每个object_type的id和，即select object_type,sum(id) from t group by object_type

新加入了LINESKIP，在map阶段用来统计出错的行；运行阶段加入了运行耗时

点击(此处)折叠或打开

  1. package exhadp;

  3. import java.io.IOException;
  4. import java.text.DateFormat;
  5. import java.text.SimpleDateFormat;
  6. import java.util.Date;

  8. import org.apache.hadoop.conf.Configuration;
  9. import org.apache.hadoop.conf.Configured;
  10. import org.apache.hadoop.fs.Path;
  11. import org.apache.hadoop.io.Text;
  12. import org.apache.hadoop.mapreduce.Job;
  13. import org.apache.hadoop.mapreduce.Mapper;
  14. import org.apache.hadoop.mapreduce.Reducer;
  15. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  16. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  17. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  18. import org.apache.hadoop.util.Tool;
  19. import org.apache.hadoop.util.ToolRunner;

  21. public class getsum extends Configured implements Tool {

  23.     enum Counter {
  24.         LINESKIP, // 出错的行
  25.     }

  27.     public int run(String[] args) throws Exception {
  28.         Configuration conf = new Configuration();

  30.         Job job = new Job(conf, "getsum"); // 任务名
  31.         job.setJarByClass(getsum.class); // 指定Class

  33.         FileInputFormat.addInputPath(job, new Path(args[0])); // 输入路径
  34.         FileOutputFormat.setOutputPath(job, new Path(args[1])); // 输出路径

  36.         job.setMapperClass(Map.class); // 调用上面Map类作为Map任务代码
  37.         job.setReducerClass(Reduce.class); // 调用上面Reduce类作为Reduce任务代码
  38.         job.setOutputFormatClass(TextOutputFormat.class);
  39.         job.setOutputKeyClass(Text.class); // 指定输出的KEY的格式
  40.         job.setOutputValueClass(Text.class); // 指定输出的VALUE的格式

  42.         job.waitForCompletion(true);

  44.         // 输出任务完成情况
  45.         System.out.println("任务名称：" + job.getJobName());
  46.         System.out.println("任务成功：" + (job.isSuccessful() ? "是" : "否"));
  47.         System.out.println("输入行数："
  48.                 + job.getCounters()
  49.                         .findCounter("org.apache.hadoop.mapred.Task$Counter",
  50.                                 "MAP_INPUT_RECORDS").getValue());
  51.         System.out.println("输出行数："
  52.                 + job.getCounters()
  53.                         .findCounter("org.apache.hadoop.mapred.Task$Counter",
  54.                                 "MAP_OUTPUT_RECORDS").getValue());
  55.         System.out.println("跳过的行："
  56.                 + job.getCounters().findCounter(Counter.LINESKIP).getValue());

  58.         return job.isSuccessful() ? 0 : 1;
  59.     }

  61.     /**
  62.      * 设置系统说明 设置MapReduce任务
  63.      */
  64.     public static void main(String[] args) throws Exception {

  66.         // 判断参数个数是否正确
  67.         // 如果无参数运行则显示以作程序说明
  68.         if (args.length != 2) {
  69.             System.err.println("");
  70.             System.err.println("Usage: getsum < input path > < output path > ");
  71.             System.err
  72.                     .println("Example: hadoop jar exhadp.jar hdfs://localhost:9000/home/james/Test_2 hdfs://localhost:9000/home/james/output");
  73.             System.err.println("Counter:");
  74.             System.err.println("\t" + "LINESKIP" + "\t"
  75.                     + "Lines which are too short");
  76.             System.exit(-1);
  77.         }

  79.         // 记录开始时间
  80.         DateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  81.         Date start = new Date();

  83.         // 运行任务
  84.         int res = ToolRunner.run(new Configuration(), new getsum(), args);

  86.         // 输出任务耗时
  87.         Date end = new Date();
  88.         float time = (float) ((end.getTime() - start.getTime()) / 60000.0);
  89.         System.out.println("任务开始：" + formatter.format(start));
  90.         System.out.println("任务结束：" + formatter.format(end));
  91.         System.out.println("任务耗时：" + String.valueOf(time) + " 分钟");

  93.         System.exit(res);
  94.     }

  96.     public static class Reduce extends Reducer<Text, Text, Text, Text> {
  97.         public void reduce(Text key, Iterable<Text> values, Context context)
  98.                 throws IOException, InterruptedException {

  100.             int sum = 0;

  102.             for (Text value : values) {
  103.                 sum = sum + Integer.parseInt(value.toString());
  104.             }

  106.             context.write(key, new Text(Integer.toString(sum)));
  107.         }
  108.     }

  110.     public static class Map extends Mapper<Object, Text, Text, Text> {
  111.         public void map(Object key, Text value, Context context)
  112.                 throws IOException, InterruptedException {
  113.             String line = value.toString(); // 读取源数据
  114.             String[] strarray = line.split(",");
  115.             try {
  116.                 Text status = new Text(strarray[6]);
  117.                 Text id = new Text(strarray[0]);

  119.                 context.write(status, id);
  120.             } catch (java.lang.ArrayIndexOutOfBoundsException e) {
  121.                 context.getCounter(Counter.LINESKIP).increment(1); // 出错令计数器+1
  122.                 return;
  123.             }
  124.         }
  125.     }
  126. }


2.7导出并运行jar包

导出jar包







将生成的jar文件复制到hadoop集群中即可运行：

[grid@dehadp01 ~]$ hadoop jar exhadp.jar getsum TEST_OBJECTS/part-m-00000 getsum_out

15/07/30 16:30:01 WARN mapreduce.JobSubmitter: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.

15/07/30 16:30:01 INFO input.FileInputFormat: Total input paths to process : 1

15/07/30 16:30:01 INFO mapreduce.JobSubmitter: number of splits:1

15/07/30 16:30:02 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1437633633589_0067

15/07/30 16:30:02 INFO impl.YarnClientImpl: Submitted application application_1437633633589_0067

15/07/30 16:30:02 INFO mapreduce.Job: The url to track the job: http://dehadp01:8088/proxy/application_1437633633589_0067/

15/07/30 16:30:02 INFO mapreduce.Job: Running job: job_1437633633589_0067

15/07/30 16:30:11 INFO mapreduce.Job: Job job_1437633633589_0067 running in uber mode : false

15/07/30 16:30:11 INFO mapreduce.Job:  map 0% reduce 0%

15/07/30 16:30:20 INFO mapreduce.Job:  map 100% reduce 0%

15/07/30 16:30:29 INFO mapreduce.Job:  map 100% reduce 100%

15/07/30 16:30:29 INFO mapreduce.Job: Job job_1437633633589_0067 completed successfully

15/07/30 16:30:29 INFO mapreduce.Job: Counters: 49

File System Counters

FILE: Number of bytes read=1253658

FILE: Number of bytes written=2723009

FILE: Number of read operations=0

FILE: Number of large read operations=0

FILE: Number of write operations=0

HDFS: Number of bytes read=10921559

HDFS: Number of bytes written=785

HDFS: Number of read operations=6

HDFS: Number of large read operations=0

HDFS: Number of write operations=2

Job Counters 

Launched map tasks=1

Launched reduce tasks=1

Data-local map tasks=1

Total time spent by all maps in occupied slots (ms)=6421

Total time spent by all reduces in occupied slots (ms)=6858

Total time spent by all map tasks (ms)=6421

Total time spent by all reduce tasks (ms)=6858

Total vcore-seconds taken by all map tasks=6421

Total vcore-seconds taken by all reduce tasks=6858

Total megabyte-seconds taken by all map tasks=6575104

Total megabyte-seconds taken by all reduce tasks=7022592

Map-Reduce Framework

Map input records=75942

Map output records=75942

Map output bytes=1101768

Map output materialized bytes=1253658

Input split bytes=115

Combine input records=0

Combine output records=0

Reduce input groups=44

Reduce shuffle bytes=1253658

Reduce input records=75942

Reduce output records=44

Spilled Records=151884

Shuffled Maps =1

Failed Shuffles=0

Merged Map outputs=1

GC time elapsed (ms)=105

CPU time spent (ms)=8210

Physical memory (bytes) snapshot=444616704

Virtual memory (bytes) snapshot=2427170816

Total committed heap usage (bytes)=306708480

Shuffle Errors

BAD_ID=0

CONNECTION=0

IO_ERROR=0

WRONG_LENGTH=0

WRONG_MAP=0

WRONG_REDUCE=0

File Input Format Counters 

Bytes Read=10921444

File Output Format Counters 

Bytes Written=785

任务名称：getsum

任务成功：是

输入行数：75942

输出行数：75942

跳过的行：0

任务开始：2015-07-30 16:29:58

任务结束：2015-07-30 16:30:29

任务耗时：0.52346665 分钟

[grid@dehadp01 ~]$ hadoop fs -cat getsum_out/part-r-00000

CLUSTER 2607

CONSUMER GROUP 309547

CONTEXT 233362

DESTINATION 24817

DIRECTORY 84148

EDITION 99

EVALUATION CONTEXT 342371

FUNCTION 14181134

INDEX 256715660

INDEX PARTITION 18627058

INDEXTYPE 524538

JAVA CLASS 867243753

JAVA DATA 17401769

JAVA RESOURCE 44179866

JAVA SOURCE 111708

JOB 315526

JOB CLASS 220990

LIBRARY 4042643

LOB 68668545

LOB PARTITION 11075

MATERIALIZED VIEW 68788

OPERATOR 2573501

PACKAGE 52912299

PACKAGE BODY 52093978

PROCEDURE 4944329

PROGRAM 245410

QUEUE 1207189

RESOURCE PLAN 123925

RULE 12864

RULE SET 471244

SCHEDULE 37694

SCHEDULER GROUP 50096

SEQUENCE 6181131

SYNONYM 993156532

TABLE 124894005

TABLE PARTITION 75556978

TABLE SUBPARTITION 281680

TRIGGER 43385559

TYPE 110249150

TYPE BODY 9475369

UNDEFINED 136498

VIEW 109084203

WINDOW 112671

XML SCHEMA 3135344


来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/20777547/viewspace-1760222/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： hadoop2.6 HA集群安装配置
下一篇： 学习HBASE
请登录后发表评论 登录
全部评论
paololiu

注册时间：2013-10-17

  • 博文量
    16
  • 访问量
    127043

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593