[hadoop in Action] 第3章 Hadoop组件

时间 2019-12-11

标签 hadoop action 组件栏目 Hadoop 繁體版

原文原文链接

管理HDFS中的文件
分析MapReduce框架中的组件
读写输入输出数据

一、HDFS文件操做

［命令行方式］

Hadoop的文件命令采起的形式为：

hadoop fs -cmd <args>

其中，cmd是具体的文件命令，而<args>是一组数目可变的参数。

（1）添加文件和目录

HDFS有一个默认的工做目录/user/$USER，其中$USER是你的登陆用户名。不过这个目录不会自动创建，让咱们用mkdir命令建立它。Hadoop的mkdir命令会自动建立父目录，相似于UNIX中使用-p选项的mkdir命令。

hadoop fs -mkdir /user/chuck

若是想看到全部的子目录，则可使用hadoop的lsr命令，相似于UNIX中打开－r选项的ls：

hadoop fs -lsr /

［输出结果显示出属性信息，好比权限、全部者、组、文件大小以及最后修改日期，全部这些都相似于UNIX的概念。显示“1”的列给出文件的复制因子。由于复制因子不适用于目录，故届时该列仅会显示一个破折号(-)］

在本地文件系统中建立一个名为examle.txt的文本文件，用hadoop的put命令将它从本地文件系统复制到HDFS中：

hadoop fs -put example.txt ./

（2）获取文件

从HDFS中复制文件到本地文件系统：

hadoop fs -get example.txt ./

显示HDFS中文件的内容：

hadoop fs -cat example.txt

［能够在hadoop的文件命令中使用UNIX的管道，将其结果发送给其余的UNIX命令作进一步处理］

查看最后一千字节：

hadoop fs -tail example.txt

（3）删除文件

删除HDFS中的文件：

hadoop fs -rm example.txt

［ rm命令还能够用于删除空目录］

hadoop fs -rmr /user/chuck

（4）查阅帮助

hadoop fs -help <cmd>

［编程方式］

hadoop命令行工具中有一个getmerge命令，用于把一组HDFS文件在复制到本地计算机之前进行合并，下面开发的是实现把本地计算机文件复制到HDFS之前进行合并：

代码清单 PutMerge程序

 1 import java.io.IOException;
 2 
 3 import org.apache.hadoop.conf.Configuration;
 4 import org.apache.hadoop.fs.FSDataInputStream;
 5 import org.apache.hadoop.fs.FSDataOutputStream;
 6 import org.apache.hadoop.fs.FileStatus;
 7 import org.apache.hadoop.fs.FileSystem;
 8 import org.apache.hadoop.fs.Path;
 9 
10 public class PutMerge {
11 
12     public static void main(String[] args) throws IOException {
13 
14         Configuration conf = new Configuration();
15         FileSystem hdfs  = FileSystem.get(conf);
16         FileSystem local = FileSystem.getLocal(conf);
17 
18         Path inputDir = new Path(args[0]);   //（1）设定输入目录和输出文件
19         Path hdfsFile = new Path(args[1]); 
20 
21         try {
22             FileStatus[] inputFiles = local.listStatus(inputDir);    //（2）获得本地文件列表
23             FSDataOutputStream out = hdfs.create(hdfsFile);    //（3）生成HDFS输出流
24 
25             for (int i=0; i<inputFiles.length; i++) {
26                 System.out.println(inputFiles[i].getPath().getName());
27                 FSDataInputStream in = local.open(inputFiles[i].getPath());    //（4）打开本地输入流
28                 byte buffer[] = new byte[256];
29                 int bytesRead = 0;
30                 while( (bytesRead = in.read(buffer)) > 0) {
31                     out.write(buffer, 0, bytesRead);
32                 }
33                 in.close();
34             }
35             out.close();
36         } catch (IOException e) {
37             e.printStackTrace();
38         }
39     }
40 }

（1）根据用户定义的参数设置本地目录和HDFS的目标文件；

（2）提取本地输入目录中每一个文件的信息；

（3）建立一个输出流写入到HDFS文件；

（4）遍历本地目录中的每一个文件，打开一个输入流来读取该文件

FileSystem类还有些方法用于其余标准文件操做，如delete()、exists()、mkdirs()和rename()。

二、剖析MapReduce程序

MapReduce程序经过操做键/值对来处理数据，通常形式为：

map：(k1, v1) ——> list(k2, v2)

reduce：(k2, list(v2)) ——> list(k3,v3)

输入数据；
输入数据被分布在节点上；
每一个map任务处理一个数据分片；
Mapper输出中间数据；
节点间的数据交换在“洗牌”阶段完成；
相同key的中间数据进入相同的reducer；
存储Reducer的输出。

虽然咱们能够而且的确常常把某些键与值称为整数、字符串等，但它们实际上并非Integer、String等那些标准的Java类。这是由于为了让键/值对能够在集群上移动，MapReduce框架提供了一种序列化键/值对的方法。所以，只有那些支持这种序列化的类可以在这个框架中充当键或者值。

更具体而言，实现Writable接口的类能够是值，而实现WritableComparable<T>接口的类既能够是键也能够是值。注意WritableComparable<T>接口是Writable和java.lang.Comparable<T>接口的组合。对于键而言，咱们须要这个比较，由于它们将在Reduce阶段进行排序，而值仅会被简单地传递。

Hadoop带有一些预约义的类用于实现WritableComparable，包括面向全部基本数据类型的封装类，以下表：

类	描述
BooleanWritable	标准布尔变量的封装
ByteWritable	单字节数的封装
DoubleWritable	双字节数的封装
FloatWritable	浮点数的封装
IntWritable	整数的封装
LongWritable	长整数的封装
Text	使用UTF8格式的文本封装
NullWritable	无键值的占位符

键和值所采用的数据类型能够超过Hadoop自身所支持的基本类型，能够自定义数据类型，只要它实现了Writable（或WritableComparable<T>）接口。

代码清单示例实现WritableComparable接口的类

 1 import java.io.DataInput;
 2 import java.io.DataOutput;
 3 import java.io.IOException;
 4  
 5 import org.apache.hadoop.io.WritableComparable;
 6  
 7 public class Edge implements WritableComparable<Edge> {
 8  
 9     private String departureNode;
10     private String arrivalNode;
11  
12     public String getDepartureNode() { return departureNode;}
13  
14     @Override
15     public void readFields(DataInput in) throws IOException {    //（1）说明如何读入数据
16         departureNode = in.readUTF();
17         arrivalNode = in.readUTF();     
18     }
19  
20     @Override
21     public void write(DataOutput out) throws IOException {    //（2）说明如何写入数据
22         out.writeUTF(departureNode);
23         out.writeUTF(arrivalNode); 
24     }
25  
26     @Override
27     public int compareTo(Edge o) {    //（3）定义数据排序
28      return (departureNode.compareTo(o.departureNode) != 0)
29          ? departureNode.compareTo(o.departureNode)
30          : arrivalNode.compareTo(o.arrivalNode);
31     }
32 }

这个Edge类实现了Writable接口的readFields()及write()方法。它们与Java中的DataInput和DataOutput类一块儿用于类中内容的串行化。而Comparable接口中的实现是compareTo()方法。若是被调用的Edge小于、等于或者大于给定的Edge，这个方法会分别返回－1，0，1。

[Mapper]

一个类要做为mapper，需继承MapReducebase基类并实现Mapper接口。并不奇怪，mapper和reducer的基类均为MapReduceBase类。它包含类的构造与解构方法。

void configure(JobConfjob)：该函数提取XML配置文件或者应用程序主类中的参数，在数据处理以前调用该函数。
void close()：做为map任务结束前的最后一个操做，该函数完成全部的结尾工做，如关闭数据库链接、打开文件等。

Mapper接口负责数据处理阶段。它采用的形式为Mapper<k1,v1,k2,v2>Java泛型，这里键类和值类分别实现WritableComparable和Writable接口。Mapper只有一个方法——Map，用于处理一个单独的键/值对。

void map (k1 key, v1 value, OutputCollector<k2,v2> output, Reporter reporter) throws IOException

该函数处理一个给定的键/值对 (k1,v1)，生成一个键/值对(k2,v2)的列表(该列表也可能为空)。OutputCollector接收这个映射过程的输出，Reporter能够提供对mapper相关附加信息的记录，造成任务进度。

Hadoop提供了一些有用的mapper实现，以下表：

类	描述
IdentityMapper<k,v>	实现Mapper<k,v,k,v>将输入直接映射到输出
InverseMapper<k,v>	实现Mapper<k,v,v,k>反转键/值对
RegexMapper<k>	实现Mapper<k,text,text,LongWritable>,为每一个常规表达式的匹配项生成一个(match,1)对
TokenCountMapper<k>	实现Mapper<k,text,text,LongWritable>,当输入的值为分词时，生成一个(token,1)对

[Reducer]

reducer的实现和mapper同样必须首先在MapReduce基类上扩展，容许配置和清理。此外，它还必须实现Reducer接口使其具备以下的单一方法：

void reduce(k2 key, Iterator<v2> values, OutputCollector<k3,v3> output, Reporter reporter) throws IOException

当reducer任务接收来自各个mapper的输出时，它按照键/值对中的键对输入数据进行排序，并将相同键的值归并。而后调用reduce()函数，并经过迭代处理那些与指定键相关联的值，生成一个(可能为空的)列表(k3,v3)。OutputCollector接收reduce阶段的输出，并写入输出文件。Reporter可提供对reducer相关附加信息的记录，造成任务进度。

Hadoop提供了一些基本的reducer实现，以下表：

类	描述
IdentityReudcer<k,v>	实现Reducer<k,v,k,v>将输入直接映射到输出
LongSumReducer<k>	实现<k,LongWritable,k,LongWritable>, 计算与给定键相对应的全部值的和

[Partitioner：重定向Mapper输出]

当使用多个reducer时，咱们就须要采起一些办法来肯定mapper应该把键/值对输出给谁。默认的做法是对键进行散列来肯定reducer。hadoop经过HashPartitioner类强制执行这个策略。但有时HashPartitioner会让你出错。

 1 public class EdgePartitioner implements Partitioner<Edge, Writable>
 2 {
 3      @verride
 4      public int getPartition(Edge key, Writable value, int numPartitions)
 5      {
 6           return key.getDepartureNode().hashCode() % numPartitions;
 7      }
 8  
 9      @verride
10      public void configure(JobConf conf) { }
11 }

一个定制的partitioner只须要实现configure()和getPartition()两个函数。前者将hadoop对做业的配置应用在patittioner上，然后者返回一个介于0和reducer任务数之间的整数，指向键/值对将要发送的reducer。

在map和reduce阶段之间，一个MapReduce应用必然从mapper任务获得输出结果，并把这些结果发布给reduce任务。该过程一般被称为洗牌。

[Combiner：本地reduce]

在许多MapReduce应用场景中，咱们不妨在分发mapper结果以前作一下“本地Reduce”。

[预约义的mapper和reducer类的单词计数]

代码清单修改的WordCount例程

 1 import org.apache.hadoop.fs.Path;
 2 import org.apache.hadoop.io.Text;
 3 import org.apache.hadoop.io.LongWritable;
 4 import org.apache.hadoop.mapred.FileInputFormat;
 5 import org.apache.hadoop.mapred.FileOutputFormat;
 6 import org.apache.hadoop.mapred.JobClient;
 7 import org.apache.hadoop.mapred.JobConf;
 8 import org.apache.hadoop.mapred.lib.TokenCountMapper;
 9 import org.apache.hadoop.mapred.lib.LongSumReducer;
10  
11 public class WordCount2 {
12     public static void main(String[] args) {
13         JobClient client = new JobClient();
14         JobConf conf = new JobConf(WordCount2.class);
15  
16         FileInputFormat.addInputPath(conf, new Path(args[0]));
17         FileOutputFormat.setOutputPath(conf, new Path(args[1]));
18  
19         conf.setOutputKeyClass(Text.class);
20         conf.setOutputValueClass(LongWritable.class);
21         conf.setMapperClass(TokenCountMapper.class);
22         conf.setCombinerClass(LongSumReducer.class);
23         conf.setReducerClass(LongSumReducer.class);
24  
25         client.setConf(conf);
26         try {
27             JobClient.runJob(conf);
28         } catch (Exception e) {
29             e.printStackTrace();
30         }
31     }
32 }

三、读和写

[InputFormat]

hadoop分割与读取输入文件的方式被定义在InputFormat接口的一个实现中。TextInputFormat是InputFormat的默认实现，当你想要一次获取一行内容而输入数据又没有肯定的键值时，这种数据格式一般会很是有用。

经常使用的InputFormat类，以下表：

InputFormat	描述
TextInputFormat	在文本文件中每一行均为一个记录。键(key)为一行的字节偏移，而值(value)为一行的内容 key: LongWritable value: Text
KeyValueTextInputFormat	在文本文件中的每一行均为一个记录。以每行的第一个分隔符为界，分隔符以前的是键(key)，以后的是值(value)。分离器在属性key.value.separator.in.input.line中设定，默认为制表符(\t)。 key: Text Value: Text
SequenceFileInputFormat<k,v>	用于读取序列文件的InputFormat。键和值由用户定义。序列文件为hadoop专用的压缩二进制文件格式。它专用于一个MapReduce做业和其余MapReduce做业之间传送数据。 key: K(用户定义) value: V(用户定义)
NLineInputFormat	与TextInputFormat相同，但每一个分片必定有N行。N在属性mapred.line.input.format.linespermap中设定，默认为1. key: LongWritable value: Text

能够设置JobConf对象使用KeyValueTextInputFormat类读取这个文件：

conf.setInputFormat(KeyValueTextInputFormat.class);

回想一下，咱们以前在mapper中曾使用LongWritable和Text分别做为键(key)和值(value)的类型。在TextInputFormat中，由于值为用数字表示的偏移量，因此LongWritable是一个合理的键类型。而当使用KeyvalueTextInputFormat时，不管是键和值都为Text类型，你必须改变mapper的实现以及map()方法来适应这个新的键(key)类型。

生成一个定制的InputFormat：略

[OutputFormat]

当MapReduce输出数据到文件时，使用的是OutputForamt类，它与inputForamt类类似。由于每一个reducer仅需将它的输出写入本身的文件中，输出无需分片。输出文件放在一个公用目录中，一般命名为part-nnnnn，这里nnnnn是reducer的分区ID。RecordWriter对象将输出结果进行格式化，而RecordReader对输入格式进行解析。

经常使用的OutputFormat类，以下表：

OutputFormat	描述
TextOutputFormat<k,v>	将每一个记录写为一行文本。键和值以字符串的形式写入，并以制表符(\t)分隔。这个分隔符能够在属性mapred.textoutputformat.separator中修改
SequenceFileOutputFormat<k,v>	以hadoop专有序列文件格式写入键/值对。与SequenceFileInputForamt配合使用
NullOutputFormat<k,v>	无输出

[转载请注明] http://www.cnblogs.com/zhengrunjian/