Changes between Version 3 and Version 4 of adanalytics

Timestamp:

11/27/2012 11:38:36 AM (13 years ago)

Author:

liaojiaohe

Comment:

--

adanalytics

@@ -1 +1 @@
 比较完整的Hadoop应用jar包，包括入口程序，任务和帮助类3部分就够了
 
-1. 我们在项目的包的顶级放一个入口程序[[BR]]
+'''1. 我们在项目的包的顶级放一个入口程序[[BR]]'''
 如 cn.pconline.ad.analytics3.Analyser,入口程序里面有一个ProgramDriver,把任务类加在里面就可以通过命令'' hadoop jar Analyser3.0.jar SumPvClick'' 执行
 
@@ -16 +16 @@
 }}}
 
-2. hadoop任务是程序包的核心，借用网上两张图说明一下map和reduce的流程[[BR]]
+'''2. hadoop任务是程序包的核心'''[[BR]]
 我们的做法是写一个实现Tool接口的基类，其他的任务继承这个基类，有些公共的方法就不用每个重复
 {{{
 public abstract class JobBase implements Tool
+
 }}}
+
+在每个job里增加main函数，使用 ToolRunner.run 方法调用，这样有些系统的参数如map数量，系统就会先保存到conf里面[[BR]]
+传给run的arg都是经过GenericOptionsParser过虑的，在run的方法里面我们可以配置任务各部分需要的类
+{{{
+
+    public static int main(String[] args) throws Exception {
+        return ToolRunner.run(new Configuration(), new RefererUniq(), args);
+    }
+
+@Override  
+public int run(String[] args) throws Exception {  
+        Job job = new Job(conf);
+        job.setJarByClass(this.getClass());
+        job.setJobName(jobName);
+
+        job.setOutputKeyClass(Text.class);
+        job.setOutputValueClass(IntWritable.class);
+        。。。。
+
+        FileOutputFormat.setOutputPath(job, new Path(userArgs[ARGNAME.WORK.ordinal()]));
+
+        job.setMapperClass(MyMapper.class);
+        job.setCombinerClass(IntSumReducer.class);
+        job.setReducerClass(IntSumReducer.class);
+        boolean success = job.waitForCompletion(true);
+}
+}}}
+提醒一下新的hadoop mapreduce类在包 org.apache.hadoop.mapreduce.*下[[BR]]
+
 
 每个任务的核心是map和reduce，有些类可以使用系统现有的，有些可以项目公用，不过更多的需要每个job里面自己定制，借网上的两张图说明一下map和reduce的流程
 [[Image(map.jpg)]]
 [[Image(reduce.jpg)]]
+
+InputFormat做Map前的预处理，主要负责以下工作：
+验证输入的格式是否符合Configuration的输入定义，这个在实现Map和构建Conf的时候就会知道，不定义可以是Writable的任意子类。[[BR]]
+
+将input的文件切分为逻辑上的输入InputSplit，其实这就是在上面提到的在分布式文件系统中blocksize是有大小限制的，因此大文件会被划分为多个block。
+通过RecordReader来再次处理inputsplit为一组records，输出给Map。[[BR]]
+RecordReader处理后的结果作为Map的输入，Map执行定义的Map逻辑，输出处理后的key和value对应到临时中间文件。[[BR]]
+更多的输入输例子可以看[/wiki/inputouput/ Hadoop常用输入输出][[BR]]
+
+
+Combiner可选择配置，主要作用是在每一个Map执行完分析以后，在本地优先作Reduce的工作，减少在Reduce过程中的数据传输量。[[BR]]
+我们在广告统计pv的时候使用了，这样可以减少总体排序的数量和网络上数据传输量[[BR]]
+CombinerClass可以和reduce类相同也可以不同
+{{{
+        job.setMapperClass(MyMapper.class);
+        job.setCombinerClass(IntSumReducer.class);
+        job.setReducerClass(IntSumReducer.class);
+}}}
+
+
+Partitioner可选择配置，主要作用是在多个Reduce的情况下，指定Map的结果由某一个Reduce处理，每一个Reduce都会有单独的输出文件。[[BR]]
+主要使用场景是想一次读文件，得到多样的结果，比如读广告的原始文件，希望一次得到所有唯一的ip和唯一的uv可以使用这种[[BR]]
+但如果搞得太复杂会影响程序的可读性，增加判断肯定会影响job的速度，有时候我宁愿多读一次输入。[[BR]]
+
+
+Reduce执行具体的业务逻辑，并且将处理结果输出给OutputFormat。[[BR]]
+一个简单的reduce里面，要注意的是reduce的迭代器只能读取一次Iterable<IntWritable> values，如果你要遍历所有values，要注意会不会OOM。
+{{{
+    /**
+     * 累计value的reducer类
+     * @param <K> 
+     */
+    public static class IntSumReducer<K>
+            extends Reducer<K, IntWritable, K, IntWritable> {
+
+        private IntWritable result = new IntWritable();
+
+        @Override
+        public void reduce(K key, Iterable<IntWritable> values,
+                Context context) throws IOException, InterruptedException {
+            int sum = 0;
+            for (IntWritable val : values) {
+                sum += val.get();
+            }
+            result.set(sum);
+            context.write(key, result);
+        }
+    }
+}}}
+
+
+OutputFormat的职责是，验证输出目录是否已经存在，同时验证输出结果类型是否如Config中配置，最后输出Reduce汇总后的结果。[[BR]]
+除非你要改变hadoop的输出规则，否则建议不需要自己写这部分，使用系统的就可以了，我们曾经试图把输出写到同一个文件上，这样hadoop只能使用单个的reduce
+完成job的时间也是极慢的。
+更多的输入输例子可以看[/wiki/inputouput/ Hadoop常用输入输出][[BR]]