Hadoop实战读书笔记（8）

什么是开发数据集？

一个流行的开发策略是为生产环境中的大数据集创建一个较小的、抽样的数据子集，称为开发数据集。这个开发数据集可能只有几百兆字节。当你以单机或者伪分布式模式编写程序来处理它们时，你会发现开发周期很短，在本身的机器上运行程序也很方便，并且还能够在独立的环境中进行调试。

 

为何选择专利引用数据作测试？

1、由于它们与你未来会遇到的大多数数据类型类似

2、专利引用数据所构成的关系图与网页连接以及社会网络图可谓大同小异

3、专利发布以时间为序，有些特性相似于时间序列

4、每一个专利关联到一我的 (发明人) 和一个位置 (发明人的国家)，你能够将之视为我的信息或地理数据

5、你能够将这些数据视为具备明确模式的普通数据库关系，而格式上简单地以逗号分开

 

数据集采用标准

数据集采用标准的逗号分隔取值 (comma-separated values, CSV) 格式。

 

构建MapReduce程序的基础模板

大多数MapReduce程序的编写均可以简单地依赖于一个模板及其变种，当撰写一个新得MapReduce程序时，咱们一般会采用一个现有的MapReduce程序，并将其修改为咱们所但愿的样子。

 

典型的Hadoop程序的模板

public class MyJob extends Configured implements Tool {

       public static class MapClass extends MapReduceBase

              implements Mapper<Text, Text, Text, Text> {

              public void map (Text key, Text value,

                                   OutputCollector<Text, Text> output,

                                   Reporter reporter) throws IOException {

                     output.collect(value, key);

              }

       }

       public static class Reduce extends MapReduceBase

              implements Reducer<Text, Text, Text, Text> {

              public void reduce(Text key, Iterator<Text> values,

                                   OutputCollector<Text, Text> output,

                                   Reporter reporter) throws IOException {

                     String csv = "";

                     while (values.hasNext()) {

                            if (csv.length() > 0) csv += ",";

                            csv += values.next().toString();

                     }

                     output.collect(key, new Text(csv));

              }

       }

       public int run(String[] args) throws Exception {

              Configuration conf = getConf();

 

              JobConf job = new JobConf(conf, MyJob.class);

 

              Path in = new Path(args[0]);

              Path out = new Path(args[1]);

              FileInputFormat.setInputPaths(job, in);

              FileOutputFormat.setOutputPath(job, out);

             

              job.setJobName("MyJob");

              job.setMapperClass(MapClass.class);

              job.setReducerClass(Reduce.class);

 

              job.setInputFormat(KeyValueTextInputFormat.class);

              job.setOutputFormat(TextOutputFormat.class);

              job.setOutputKeyClass(Text.class);

              job.setOutputValueClass(Text.class);

              job.set("key.value.separator.in.input.line", ",");

 

              JobClient.runJob(job);

 

              return 0;

       }

 

       public static void main(String[] args) throws Exception {

              int res = ToolRunner.run(new Configuration(), new MyJob(), args);

              System.exit(res);

       }

}

1、咱们习惯用单个类来完整地定义每一个MapReduce做业，这里成为MyJob类

2、Hadoop要求Mapper和Reducer必须是它们自身的静态类，这些类很是小，模板将它们包含在MyJob类中做为内部类，这样作的好处是能够把全部的东西放在一个文件内，简化代码管理

3、可是须要记住这些内部类是独立的，一般不与MyJob类进行交互

4、在做业执行期间，采用不一样JVM的各种节点复制并运行Mapper和Reducer，而其余的做业类仅在客户机上执行

 

解释下run()方法

1、框架的核心在run()方法中，也称为driver

2、它实例化、配置并传递一个JobConf对象命名的做业给JobClient.runJob()以启动MapReduce做业(反过来，JobClient类与JobTracker通讯让该做业在集群上启动)

3、JobConf对象将保持做业运行所需的所有配置参数

4、Driver须要在做业中为每一个做业定制基本参数，包括输入路径、输出路径、Mapper类和Reducer类

5、每一个做业能够重置默认的做业属性，例如，InputFormat、OutputFormat等，也能够调用JobConf对象中的set()方法填充任意的配置参数

6、一旦传递JobConf对象到JobClient.runJob()，他就被视为决定这个做业如何运行的蓝本

 

关于driver的配置的一些说明

1、JobConf对象有许多参数，但咱们并不但愿所有的参数都经过编写driver来设置，能够把Hadoop安装时的配置文件做为一个很好的起点

2、用户可能但愿在命令行启动一个做业时传递额外的参数来改变做业配置

3、Driver能够经过自定义一组命令并自行处理用户参数，来支持用户修改其中的一些配置

4、由于常常须要作这样的任务，Hadoop框架便提供了ToolRunner、Tool和Configured来简化其实现。

5、当它们在上面的MyJob框架中被同时使用时，这些类使得做业能够理解用户提供的被GenericOptionParser支持的选项

 

好比下面的命令：

bin/hadoop jar playgroup/MyJob.jar MyJob input/cite75-99.txt output

若是咱们运行做业仅仅是想看到mapper的输出 (处于调试的目的), 能够用选项 -D mapred.reduce.tasks=0将reducer的数目设置为0

bin/hadoop jar playgroup/MyJob.jar MyJob -D mapred.reduce.tasks=0 input/cite75-99.txt output

 

经过使用ToolRunner、MyJob能够自动支持一下选项
GenericOptionsParser支持的选项

选项	描述
-conf <configuration file>	指定一个配置文件
-D <property=value>	给JobConf属性赋值
-fs <local | namenode:port>	指定一个NameNode，能够是 "local"
-jt <local | jobtracker:port>	指定一个JobTracker
-files <list of files>	指定一个以逗号分隔的文件列表，用于MapReduce做业。这些文件自动地分布到全部节点，使之可从本地获取
-libjars <list of jars>	指定一个以逗号分隔的jar文件，使之包含在全部任务JVM的classpath中
-archives <list of archives>	指定一个以逗号分隔的存档文件列表，使之能够在全部任务节点上打开

 

模板代码Mappper与Reducer

模板中习惯将Mapper类称为MapClass，而将Reducer类称为Reduce

Mapper和Reducer都是MapReduceBase的扩展

MapReduceBase是个小类，包含configure()和close()，咱们使用上两个方法来创建和清除map(reduce)任务，除非是更高级的做业，一般咱们并不须要覆盖它们

 

Mapper类和Reducer类模板说明

public static class MapClass extends MapReduceBase

       implements Mapper<K1, V1, K2, V2> {

       public void map (K1 key, V1 value,

                                   OutputCollector<K2, V2> output,

                                   Reporter reporter) throws IOException { }

}

public static class Reduce extends MapReduceBase

       implements Reducer<K1, V2, K3, V3> {

       public void reduce(K2 key, Iterator<V2> values,

                                   OutputCollector<K3, V3> output,

                                   Reporter reporter) throws IOException { }

}

Mapper类的核心操做为map()方法，Reduce类为reduce()方法。每个map()方法的调用分别被赋予一个类型为K1和V1的键/值对。这个键/值对由mapper生成，并经过OutputCollector对象的collect()方法来输出。你须要在map()方法中的合适位置调用:

output.collect((K2) k, (V2) v);

 

在Reducer中reduce()方法的每次调用均被赋予K2类型的键，以及V2类型的一组值。注意它必须与Mapper中使用的K2和V2类型相同。Reduce()方法可能会循环遍历V2类型的全部值。

while (values.hasNext()) {

       V2 v = values.next();

}

Reduce()方法还使用OutputCollector来搜集其键/值的输出，它们的类型为K3/V3。在reudce()方法中能够调用

output.collect((K3) k, (V3) v);

除了在Mapper和Reducer之间保持K2与V3的类型一致，还须要确保在Mapper和Reducer中使用的键值类型与在driver中设置的输入格式、输出键的类，以及输出值的类保持一致

使用KeyValueTextInputFormat意味着K1和V1必须均为Text类型

Driver则必须调用setOutputKeyClass()和setOutputValueClass()分别指定K2和V2的类

最终：

1、全部的键与值的类型必须是Writable的子类型，来确保Hadoop的序列化接口能够把数据在分布式集群上发送

2、键的类型实现了WritableComparable，它是Writable的子接口，键的类型还需额外支持compareTo()方法，由于在MapReduce框架中键会被用来进行排序